API Reference

This section contains the complete API reference for JACTUS.

Core Modules

Core Types

Time and DateTime

Date and time handling for ACTUS contracts.

This module provides the ActusDateTime class and utilities for parsing, manipulating, and comparing dates according to ACTUS specifications.

Key features: - ISO 8601 datetime parsing with ACTUS-specific extensions - Support for 24:00:00 (end of day) and 00:00:00 (start of day) - Period/cycle arithmetic (e.g., adding ‘3M’ to a date) - Month-end handling and leap year support - JAX pytree registration for functional programming - Business day awareness

References

ACTUS Technical Specification v1.1, Section 3 (Time)

class jactus.core.time.ActusDateTime(year, month, day, hour=0, minute=0, second=0)[source]

Bases: object

Immutable datetime representation for ACTUS contracts.

ACTUS uses ISO 8601 datetime strings with special handling: - 24:00:00 represents end of day (midnight of next day) - 00:00:00 represents start of day (midnight) - Dates can be added/subtracted using cycle notation (e.g., ‘3M’, ‘1Y’)

Parameters:
year

Year (1-9999)

Type:

int

month

Month (1-12)

Type:

int

day

Day of month (1-31)

Type:

int

hour

Hour (0-24, where 24 = end of day)

Type:

int

minute

Minute (0-59)

Type:

int

second

Second (0-59)

Type:

int

Example

>>> dt = ActusDateTime.from_iso("2024-01-15T00:00:00")
>>> dt.add_period("3M")
ActusDateTime(2024, 4, 15, 0, 0, 0)

References

ACTUS Technical Specification v1.1, Section 3.1

year: int
month: int
day: int
hour: int = 0
minute: int = 0
second: int = 0
__post_init__()[source]

Validate datetime components.

Return type:

None

classmethod from_iso(iso_string)[source]

Parse ISO 8601 datetime string.

Supports formats: - YYYY-MM-DD - YYYY-MM-DDTHH:MM:SS - YYYY-MM-DD HH:MM:SS (space separator)

Special handling for 24:00:00 (end of day).

Parameters:

iso_string (str) – ISO 8601 formatted datetime string

Returns:

ActusDateTime instance

Raises:

ValueError – If string format is invalid

Return type:

ActusDateTime

Example

>>> ActusDateTime.from_iso("2024-01-15T24:00:00")
ActusDateTime(2024, 1, 15, 24, 0, 0)
to_iso()[source]

Convert to ISO 8601 string.

Returns:

MM:SS)

Return type:

ISO 8601 formatted string (YYYY-MM-DDTHH

Example

>>> dt = ActusDateTime(2024, 1, 15, 12, 30, 0)
>>> dt.to_iso()
'2024-01-15T12:30:00'
to_datetime()[source]

Convert to Python datetime object.

Note: 24:00:00 is converted to 00:00:00 of the next day.

Returns:

Python datetime object

Return type:

datetime

Example

>>> dt = ActusDateTime(2024, 1, 15, 24, 0, 0)
>>> dt.to_datetime()
datetime(2024, 1, 16, 0, 0, 0)
add_period(cycle, end_of_month_convention=EndOfMonthConvention.SD)[source]

Add a period to this datetime.

Periods are specified in ACTUS cycle notation: - NPS where N=number, P=period type, S=stub indicator - Period types: D=days, W=weeks, M=months, Q=quarters, H=half-years, Y=years - Stub indicator: ‘-’ for short stub, ‘+’ for long stub (optional)

Parameters:

  • cycle (str) – Period to add (e.g., ‘3M’, ‘1Y’, ‘2W’)

  • end_of_month_convention (EndOfMonthConvention) – How to handle month-end dates

Returns:

New ActusDateTime after adding period

Return type:

ActusDateTime

Example

>>> dt = ActusDateTime(2024, 1, 31, 0, 0, 0)
>>> dt.add_period("1M", EndOfMonthConvention.EOM)
ActusDateTime(2024, 2, 29, 0, 0, 0)  # Leap year

References

ACTUS Technical Specification v1.1, Section 3.2

is_end_of_month()[source]

Check if this date is the last day of the month.

Returns:

True if this is the last day of the month

Return type:

bool

Example

>>> ActusDateTime(2024, 2, 29, 0, 0, 0).is_end_of_month()
True
>>> ActusDateTime(2024, 2, 28, 0, 0, 0).is_end_of_month()
False
days_between(other)[source]

Calculate actual days between this date and another.

Parameters:

other (ActusDateTime) – Other datetime

Returns:

Number of days (can be negative if other is earlier)

Return type:

int

Example

>>> dt1 = ActusDateTime(2024, 1, 15, 0, 0, 0)
>>> dt2 = ActusDateTime(2024, 1, 18, 0, 0, 0)
>>> dt1.days_between(dt2)
3
years_between(other)[source]

Calculate approximate years between dates (actual days / 365.25).

Parameters:

other (ActusDateTime) – Other datetime

Returns:

Approximate years (can be negative)

Return type:

float

Example

>>> dt1 = ActusDateTime(2024, 1, 15, 0, 0, 0)
>>> dt2 = ActusDateTime(2025, 1, 15, 0, 0, 0)
>>> abs(dt1.years_between(dt2) - 1.0) < 0.01
True
__eq__(other)[source]

Check equality with another ActusDateTime.

Parameters:

other (object)

Return type:

bool

__lt__(other)[source]

Check if this datetime is before another.

Parameters:

other (ActusDateTime)

Return type:

bool

__le__(other)[source]

Check if this datetime is before or equal to another.

Parameters:

other (ActusDateTime)

Return type:

bool

__gt__(other)[source]

Check if this datetime is after another.

Parameters:

other (ActusDateTime)

Return type:

bool

__ge__(other)[source]

Check if this datetime is after or equal to another.

Parameters:

other (ActusDateTime)

Return type:

bool

__hash__()[source]

Hash for use in dicts/sets.

Return type:

int

__init__(year, month, day, hour=0, minute=0, second=0)
Parameters:
Return type:

None

jactus.core.time.parse_iso_datetime(iso_string)[source]

Parse ISO 8601 datetime string into ActusDateTime.

Supports formats: - YYYY-MM-DD - YYYY-MM-DDTHH:MM:SS - YYYY-MM-DD HH:MM:SS

Parameters:

iso_string (str) – ISO 8601 formatted string

Returns:

ActusDateTime instance

Raises:

ValueError – If format is invalid

Return type:

ActusDateTime

Example

>>> parse_iso_datetime("2024-01-15T12:30:00")
ActusDateTime(2024, 1, 15, 12, 30, 0)
jactus.core.time.parse_cycle(cycle)[source]

Parse ACTUS cycle notation.

Cycle format: NPS - N: number (integer) - P: period type (D/W/M/Q/H/Y) - S: stub indicator (‘-’ short, ‘+’ long) - optional

Parameters:

cycle (str) – Cycle string (e.g., ‘3M’, ‘1Y-’, ‘6M+’)

Returns:

Tuple of (number, period_type, stub_indicator)

Raises:

ValueError – If cycle format is invalid

Return type:

tuple[int, str, str]

Example

>>> parse_cycle("3M")
(3, 'M', '')
>>> parse_cycle("1Y-")
(1, 'Y', '-')

References

ACTUS Technical Specification v1.1, Section 3.2

jactus.core.time.add_period(dt, cycle, end_of_month_convention=EndOfMonthConvention.SD)[source]

Add a period to a datetime according to ACTUS conventions.

Parameters:

  • dt (ActusDateTime) – Starting datetime

  • cycle (str) – Period to add (e.g., ‘3M’, ‘1Y’)

  • end_of_month_convention (EndOfMonthConvention) – How to handle month-end dates

Returns:

New datetime after adding period

Return type:

ActusDateTime

Example

>>> dt = ActusDateTime(2024, 1, 31, 0, 0, 0)
>>> add_period(dt, "1M", EndOfMonthConvention.EOM)
ActusDateTime(2024, 2, 29, 0, 0, 0)

References

ACTUS Technical Specification v1.1, Section 3.2, 3.3

jactus.core.time.is_business_day(dt, calendar=Calendar.MONDAY_TO_FRIDAY)[source]

Check if a date is a business day according to the given calendar.

Parameters:
Returns:

True if date is a business day

Return type:

bool

Example

>>> dt = ActusDateTime(2024, 1, 15, 0, 0, 0)  # Monday
>>> is_business_day(dt)
True

References

ACTUS Technical Specification v1.1, Section 3.4

jactus.core.time.adjust_to_business_day(dt, convention, calendar=Calendar.MONDAY_TO_FRIDAY)[source]

Adjust a date to a business day according to the given convention.

ACTUS business day conventions have two components:

  • S (Shift): Move the payment/settlement date to a business day.

  • C (Calculate): Use the original (unadjusted) date for accrual calculations, even if the payment date was shifted.

Convention naming: [S|CS][F|MF|P|MP]

  • S prefix = Shift only (both payment and calculation use shifted date)

  • CS prefix = Calculate-Shift (payment is shifted, calculation uses original date)

Currently, all conventions implement the Shift (S) semantics for date adjustment. The CS distinction is documented for callers that need to track the original calculation date separately (see get_calculation_date).

Parameters:
Returns:

Adjusted (shifted) date (may be same as input if already a business day)

Return type:

ActusDateTime

Example

>>> dt = ActusDateTime(2024, 1, 13, 0, 0, 0)  # Saturday
>>> adjust_to_business_day(dt, BusinessDayConvention.SCF)
ActusDateTime(2024, 1, 15, 0, 0, 0)  # Monday

References

ACTUS Technical Specification v1.1, Section 3.4

jactus.core.time.is_shift_calculate(convention)[source]

Check if convention is Shift-Calculate (S prefix).

In SC conventions, both the payment date and the calculation date are shifted to a business day together.

Parameters:

convention (BusinessDayConvention) – Business day convention

Returns:

True if this is a Shift-only convention (SCF, SCMF, SCP, SCMP)

Return type:

bool

jactus.core.time.is_calculate_shift(convention)[source]

Check if convention is Calculate-Shift (CS prefix).

In CS conventions, the payment date is shifted to a business day, but the calculation date remains the original (unadjusted) date. This means interest accrual uses the scheduled date, not the actual payment date.

Parameters:

convention (BusinessDayConvention) – Business day convention

Returns:

True if this is a Calculate-Shift convention (CSF, CSMF, CSP, CSMP)

Return type:

bool

jactus.core.time.get_calculation_date(original_dt, shifted_dt, convention)[source]

Get the date to use for accrual calculations.

For SC (Shift-Calculate) conventions, returns the shifted date. For CS (Calculate-Shift) conventions, returns the original date. For NULL convention, returns the original date.

Parameters:
Returns:

Date to use for interest accrual calculations

Return type:

ActusDateTime

Contract Attributes

Contract attributes for ACTUS contracts.

This module provides the ContractAttributes class, which represents all possible attributes (contract terms) for ACTUS contracts using Pydantic for validation.

References

ACTUS Technical Specification v1.1, Sections 4-5 (Contract Attributes)

class jactus.core.attributes.ContractAttributes(*, contract_id, contract_type, contract_role, status_date, contract_deal_date=None, initial_exchange_date=None, maturity_date=None, purchase_date=None, termination_date=None, analysis_dates=None, notional_principal=None, nominal_interest_rate=None, nominal_interest_rate_2=None, currency='USD', currency_2=None, notional_principal_2=None, delivery_settlement=None, settlement_date=None, option_type=None, option_strike_1=None, option_strike_2=None, option_exercise_type=None, option_exercise_end_date=None, x_day_notice=None, exercise_date=None, exercise_amount=None, settlement_period=None, delivery_type=None, contract_structure=None, future_price=None, settlement_currency=None, fixing_period=None, quantity=None, unit=None, market_object_code=None, market_object_code_of_dividends=None, dividend_cycle=None, dividend_anchor=None, day_count_convention=None, business_day_convention=BusinessDayConvention.NULL, end_of_month_convention=EndOfMonthConvention.SD, calendar=Calendar.NO_CALENDAR, interest_payment_cycle=None, interest_payment_anchor=None, interest_capitalization_end_date=None, principal_redemption_cycle=None, principal_redemption_anchor=None, fee_payment_cycle=None, fee_payment_anchor=None, rate_reset_cycle=None, rate_reset_anchor=None, scaling_index_cycle=None, scaling_index_anchor=None, next_principal_redemption_amount=None, interest_calculation_base_cycle=None, interest_calculation_base_anchor=None, array_pr_anchor=None, array_pr_cycle=None, array_pr_next=None, array_increase_decrease=None, array_ip_anchor=None, array_ip_cycle=None, array_rr_anchor=None, array_rr_cycle=None, array_rate=None, array_fixed_variable=None, rate_reset_market_object=None, rate_reset_multiplier=None, rate_reset_spread=None, rate_reset_floor=None, rate_reset_cap=None, rate_reset_next=None, fee_rate=None, fee_basis=None, fee_accrued=None, prepayment_effect=PrepaymentEffect.N, penalty_type=None, penalty_rate=None, scaling_effect=ScalingEffect.S000, scaling_index_at_status_date=None, scaling_index_at_contract_deal_date=None, scaling_market_object=None, coverage=None, credit_event_type=None, credit_enhancement_guarantee_extent=None, accrued_interest=None, interest_calculation_base=None, interest_calculation_base_amount=None, amortization_date=None, contract_performance=ContractPerformance.PF, premium_discount_at_ied=None, price_at_purchase_date=None, price_at_termination_date=None)[source]

Bases: BaseModel

All possible attributes for an ACTUS contract.

This class uses Pydantic for automatic validation. Not all attributes are required for all contract types - use validate_contract_type_compatibility() to ensure a valid configuration.

Attributes follow ACTUS naming conventions. See ACTUS documentation for the meaning and constraints of each attribute.

Example

>>> attrs = ContractAttributes(
...     contract_id="LOAN-001",
...     contract_type=ContractType.PAM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime.from_iso("2024-01-01T00:00:00"),
...     initial_exchange_date=ActusDateTime.from_iso("2024-01-15T00:00:00"),
...     maturity_date=ActusDateTime.from_iso("2029-01-15T00:00:00"),
...     notional_principal=100000.0,
...     nominal_interest_rate=0.05,
...     currency="USD"
... )

References

ACTUS Technical Specification v1.1, Section 4

Parameters:
contract_id: str
contract_type: ContractType
contract_role: ContractRole
status_date: ActusDateTime
contract_deal_date: ActusDateTime | None
initial_exchange_date: ActusDateTime | None
maturity_date: ActusDateTime | None
purchase_date: ActusDateTime | None
termination_date: ActusDateTime | None
analysis_dates: list[ActusDateTime] | None
notional_principal: float | None
nominal_interest_rate: float | None
nominal_interest_rate_2: float | None
currency: str
currency_2: str | None
notional_principal_2: float | None
delivery_settlement: str | None
settlement_date: ActusDateTime | None
option_type: str | None
option_strike_1: float | None
option_strike_2: float | None
option_exercise_type: str | None
option_exercise_end_date: ActusDateTime | None
x_day_notice: str | None
exercise_date: ActusDateTime | None
exercise_amount: float | None
settlement_period: str | None
delivery_type: str | None
contract_structure: str | None
future_price: float | None
settlement_currency: str | None
fixing_period: str | None
quantity: float | None
unit: str | None
market_object_code: str | None
market_object_code_of_dividends: str | None
dividend_cycle: Cycle | None
dividend_anchor: ActusDateTime | None
day_count_convention: DayCountConvention | None
business_day_convention: BusinessDayConvention
end_of_month_convention: EndOfMonthConvention
calendar: Calendar
interest_payment_cycle: Cycle | None
interest_payment_anchor: ActusDateTime | None
interest_capitalization_end_date: ActusDateTime | None
principal_redemption_cycle: Cycle | None
principal_redemption_anchor: ActusDateTime | None
fee_payment_cycle: Cycle | None
fee_payment_anchor: ActusDateTime | None
rate_reset_cycle: Cycle | None
rate_reset_anchor: ActusDateTime | None
scaling_index_cycle: Cycle | None
scaling_index_anchor: ActusDateTime | None
next_principal_redemption_amount: float | None
interest_calculation_base_cycle: Cycle | None
interest_calculation_base_anchor: ActusDateTime | None
array_pr_anchor: list[ActusDateTime] | None
array_pr_cycle: list[Cycle] | None
array_pr_next: list[float] | None
array_increase_decrease: list[str] | None
array_ip_anchor: list[ActusDateTime] | None
array_ip_cycle: list[Cycle] | None
array_rr_anchor: list[ActusDateTime] | None
array_rr_cycle: list[Cycle] | None
array_rate: list[float] | None
array_fixed_variable: list[str] | None
rate_reset_market_object: str | None
rate_reset_multiplier: float | None
rate_reset_spread: float | None
rate_reset_floor: float | None
rate_reset_cap: float | None
rate_reset_next: float | None
fee_rate: float | None
fee_basis: FeeBasis | None
fee_accrued: float | None
prepayment_effect: PrepaymentEffect
penalty_type: str | None
penalty_rate: float | None
scaling_effect: ScalingEffect
scaling_index_at_status_date: float | None
scaling_index_at_contract_deal_date: float | None
scaling_market_object: str | None
coverage: float | None
credit_event_type: ContractPerformance | None
credit_enhancement_guarantee_extent: str | None
accrued_interest: float | None
interest_calculation_base: InterestCalculationBase | None
interest_calculation_base_amount: float | None
amortization_date: ActusDateTime | None
contract_performance: ContractPerformance
premium_discount_at_ied: float | None
price_at_purchase_date: float | None
price_at_termination_date: float | None
model_config = {'arbitrary_types_allowed': True, 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod validate_interest_rate(v)[source]

Validate that interest rate is greater than -1 (can be negative).

Parameters:

v (float | None)

Return type:

float | None

classmethod validate_notional(v)[source]

Validate that notional is non-zero if defined.

Parameters:

v (float | None)

Return type:

float | None

classmethod validate_currency(v)[source]

Validate currency code is 3 uppercase letters.

Parameters:

v (str)

Return type:

str

validate_dates()[source]

Validate date ordering constraints.

Return type:

ContractAttributes

validate_array_schedules()[source]

Validate array schedule consistency.

Return type:

ContractAttributes

get_attribute(actus_name)[source]

Get attribute value by ACTUS short name.

Parameters:

actus_name (str) – ACTUS short name (e.g., ‘IPNR’, ‘NT’, ‘MD’)

Returns:

Attribute value

Raises:

KeyError – If ACTUS name not recognized

Return type:

Any

Example

>>> attrs.get_attribute('NT')
100000.0
set_attribute(actus_name, value)[source]

Set attribute value by ACTUS short name.

Parameters:

  • actus_name (str) – ACTUS short name (e.g., ‘IPNR’, ‘NT’, ‘MD’)

  • value (Any) – New value (will be validated)

Raises:

  • KeyError – If ACTUS name not recognized

  • ValidationError – If value is invalid

Return type:

None

Example

>>> attrs.set_attribute('NT', 150000.0)
is_attribute_defined(actus_name)[source]

Check if an attribute has a non-None value.

Parameters:

actus_name (str) – ACTUS short name

Returns:

True if attribute is defined (not None)

Return type:

bool

Example

>>> attrs.is_attribute_defined('NT')
True

Contract State

Contract state variables for ACTUS contracts.

This module provides the ContractState dataclass for representing the mutable state of a contract at a point in time. State variables are immutable and JAX-compatible for functional programming.

References

ACTUS Technical Specification v1.1, Section 6 (State Variables)

class jactus.core.states.ContractState(tmd, sd, nt, ipnr, ipac, feac, nsc, isc, xd=None, ipac1=None, ipac2=None, prnxt=None, ipcb=None, xa=None, prf=ContractPerformance.PF)[source]

Bases: object

Immutable contract state variables.

Represents the time-varying state of an ACTUS contract. All numerical values use JAX arrays for compatibility with automatic differentiation and JIT compilation.

State follows ACTUS naming conventions (lowercase versions of attribute names). See ACTUS documentation for detailed meaning of each state variable.

Parameters:
tmd

Maturity date state (Md)

Type:

jactus.core.time.ActusDateTime

nt

Notional principal

Type:

jax.Array

ipnr

Nominal interest rate

Type:

jax.Array

ipac

Accrued interest

Type:

jax.Array

ipac1

Accrued interest leg 1 (for swaps)

Type:

jax.Array | None

ipac2

Accrued interest leg 2 (for swaps)

Type:

jax.Array | None

feac

Accrued fees

Type:

jax.Array

nsc

Notional scaling multiplier

Type:

jax.Array

isc

Interest scaling multiplier

Type:

jax.Array

prf

Contract performance status

Type:

jactus.core.types.ContractPerformance

sd

Status date

Type:

jactus.core.time.ActusDateTime

prnxt

Next principal redemption amount

Type:

jax.Array | None

ipcb

Interest calculation base

Type:

jax.Array | None

xd

Exercise date (options/futures)

Type:

jactus.core.time.ActusDateTime | None

xa

Exercise amount (options/futures)

Type:

jax.Array | None

Example

>>> import jax.numpy as jnp
>>> state = ContractState(
...     tmd=ActusDateTime(2029, 1, 15, 0, 0, 0),
...     nt=jnp.array(100000.0),
...     ipnr=jnp.array(0.05),
...     ipac=jnp.array(0.0),
...     feac=jnp.array(0.0),
...     nsc=jnp.array(1.0),
...     isc=jnp.array(1.0),
...     prf=ContractPerformance.PF,
...     sd=ActusDateTime(2024, 1, 1, 0, 0, 0),
... )

References

ACTUS Technical Specification v1.1, Section 6

tmd: ActusDateTime
sd: ActusDateTime
nt: Array
ipnr: Array
ipac: Array
feac: Array
nsc: Array
isc: Array
xd: ActusDateTime | None = None
ipac1: Array | None = None
ipac2: Array | None = None
prnxt: Array | None = None
ipcb: Array | None = None
xa: Array | None = None
prf: ContractPerformance = 'PF'
replace(**changes)[source]

Create a new state with specified changes.

Since states are immutable, this creates a new ContractState instance with the specified fields replaced.

Parameters:

**changes (Any) – Field names and new values

Returns:

New ContractState with changes applied

Return type:

ContractState

Example

>>> new_state = state.replace(nt=jnp.array(90000.0))
>>> new_state.nt
Array(90000., dtype=float32)
to_dict()[source]

Convert to dictionary for serialization.

Returns:

Dictionary representation with all state variables

Return type:

dict[str, Any]

Example

>>> data = state.to_dict()
>>> data['nt']
100000.0
classmethod from_dict(data)[source]

Create ContractState from dictionary.

Parameters:

data (dict[str, Any]) – Dictionary with state variable values

Returns:

New ContractState instance

Return type:

ContractState

Example

>>> data = {'tmd': '2029-01-15T00:00:00', 'nt': 100000.0, ...}
>>> state = ContractState.from_dict(data)
__eq__(other)[source]

Check equality with another ContractState.

Parameters:

other (object)

Return type:

bool

__hash__()[source]

Hash for use in dicts/sets.

Return type:

int

__init__(tmd, sd, nt, ipnr, ipac, feac, nsc, isc, xd=None, ipac1=None, ipac2=None, prnxt=None, ipcb=None, xa=None, prf=ContractPerformance.PF)
Parameters:
Return type:

None

jactus.core.states.initialize_state(tmd, sd, nt=0.0, ipnr=0.0, prf=ContractPerformance.PF)[source]

Initialize a contract state with default values.

Convenience function for creating a new state with sensible defaults. All accrued amounts start at zero, scaling multipliers at 1.0.

Parameters:
Returns:

New ContractState with initialized values

Return type:

ContractState

Example

>>> state = initialize_state(
...     tmd=ActusDateTime(2029, 1, 15, 0, 0, 0),
...     sd=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     nt=100000.0,
...     ipnr=0.05,
... )

References

ACTUS Technical Specification v1.1, Section 6.2

Contract Events

Contract event structures for ACTUS contracts.

This module provides ContractEvent and EventSchedule classes for representing and managing contract events (cash flows and state transitions).

References

ACTUS Technical Specification v1.1, Section 2.5, 2.9 (Events)

class jactus.core.events.ContractEvent(event_type, event_time, payoff, currency, state_pre=None, state_post=None, sequence=0, calculation_time=None)[source]

Bases: object

Represents a single contract event.

An event is a discrete occurrence on the timeline that may generate a cash flow (payoff) and/or change the contract state.

Parameters:
event_type

Type of event (IED, IP, PR, MD, etc.)

Type:

jactus.core.types.EventType

event_time

When the event occurs (τ operator)

Type:

jactus.core.time.ActusDateTime

payoff

Cash flow amount (φ operator)

Type:

jax.Array

currency

Currency of payoff

Type:

str

state_pre

Contract state before event

Type:

jactus.core.states.ContractState | None

state_post

Contract state after event

Type:

jactus.core.states.ContractState | None

sequence

Sequence number for ordering same-time events

Type:

int

Example

>>> event = ContractEvent(
...     event_type=EventType.IP,
...     event_time=ActusDateTime(2024, 4, 15, 0, 0, 0),
...     payoff=jnp.array(1250.0),
...     currency="USD",
...     sequence=EVENT_SEQUENCE_ORDER[EventType.IP],
... )

References

ACTUS Technical Specification v1.1, Section 2.5

event_type: EventType
event_time: ActusDateTime
payoff: Array
currency: str
state_pre: ContractState | None = None
state_post: ContractState | None = None
sequence: int = 0
calculation_time: ActusDateTime | None = None
__lt__(other)[source]

Compare by time, then sequence.

Parameters:

other (ContractEvent)

Return type:

bool

__le__(other)[source]

Compare by time, then sequence.

Parameters:

other (ContractEvent)

Return type:

bool

__gt__(other)[source]

Compare by time, then sequence.

Parameters:

other (ContractEvent)

Return type:

bool

__ge__(other)[source]

Compare by time, then sequence.

Parameters:

other (ContractEvent)

Return type:

bool

__eq__(other)[source]

Check equality.

Parameters:

other (object)

Return type:

bool

__hash__()[source]

Hash for use in dicts/sets.

Return type:

int

to_dict()[source]

Convert to dictionary for serialization.

Returns:

Dictionary representation

Return type:

dict[str, Any]

classmethod from_dict(data)[source]

Create ContractEvent from dictionary.

Parameters:

data (dict[str, Any]) – Dictionary with event data

Returns:

New ContractEvent instance

Return type:

ContractEvent

__init__(event_type, event_time, payoff, currency, state_pre=None, state_post=None, sequence=0, calculation_time=None)
Parameters:
Return type:

None

class jactus.core.events.EventSchedule(events, contract_id)[source]

Bases: object

Immutable container for a sequence of events.

Represents the complete event schedule for a contract, maintaining events in chronological order.

Parameters:
events

Immutable tuple of events

Type:

tuple[jactus.core.events.ContractEvent, …]

contract_id

Associated contract identifier

Type:

str

Example

>>> events = [event1, event2, event3]
>>> schedule = EventSchedule(
...     events=tuple(sorted(events)),
...     contract_id="LOAN-001",
... )

References

ACTUS Technical Specification v1.1, Section 2.9

events: tuple[ContractEvent, ...]
contract_id: str
__len__()[source]

Return number of events.

Return type:

int

__iter__()[source]

Iterate over events.

Return type:

Any

__getitem__(index)[source]

Get event by index.

Parameters:

index (int)

Return type:

ContractEvent

add_event(event)[source]

Add an event and return new schedule.

Since schedules are immutable, this creates a new schedule with the event added in sorted order.

Parameters:

event (ContractEvent) – Event to add

Returns:

New EventSchedule with event added

Return type:

EventSchedule

Example

>>> new_schedule = schedule.add_event(new_event)
filter_by_type(event_type)[source]

Filter events by type.

Parameters:

event_type (EventType) – Type to filter for

Returns:

New EventSchedule with only matching events

Return type:

EventSchedule

Example

>>> ip_events = schedule.filter_by_type(EventType.IP)
filter_by_time_range(start, end)[source]

Filter events by time range.

Parameters:
Returns:

New EventSchedule with events in range

Return type:

EventSchedule

Example

>>> range_events = schedule.filter_by_time_range(
...     ActusDateTime(2024, 1, 1, 0, 0, 0),
...     ActusDateTime(2024, 12, 31, 0, 0, 0),
... )
merge(other)[source]

Merge with another schedule.

Combines events from both schedules and sorts by time/sequence.

Parameters:

other (EventSchedule) – Other schedule to merge

Returns:

New EventSchedule with merged events

Return type:

EventSchedule

Example

>>> combined = schedule1.merge(schedule2)
get_payoffs()[source]

Extract all payoffs as a JAX array.

Returns:

Array of payoff values

Return type:

Array

Example

>>> payoffs = schedule.get_payoffs()
get_times()[source]

Extract all event times.

Returns:

List of event times

Return type:

list[ActusDateTime]

Example

>>> times = schedule.get_times()
to_dict()[source]

Convert to dictionary for serialization.

Returns:

Dictionary representation

Return type:

dict[str, Any]

__init__(events, contract_id)
Parameters:
Return type:

None

jactus.core.events.tau(event)[source]

τ (tau) operator - Extract event time(s).

The τ operator extracts the time component of events.

Parameters:

event (ContractEvent | EventSchedule) – Single event or event schedule

Returns:

Event time(s)

Return type:

ActusDateTime | list[ActusDateTime]

Example

>>> t = tau(event)
>>> times = tau(schedule)

References

ACTUS Technical Specification v1.1, Section 2.5

jactus.core.events.phi(event)[source]

φ (phi) operator - Extract payoff(s).

The φ operator extracts the payoff (cash flow) component of events.

Parameters:

event (ContractEvent | EventSchedule) – Single event or event schedule

Returns:

Payoff(s) as JAX array

Return type:

Array

Example

>>> p = phi(event)
>>> payoffs = phi(schedule)

References

ACTUS Technical Specification v1.1, Section 2.5

jactus.core.events.sort_events(events)[source]

Sort events by time, then by sequence.

Parameters:

events (list[ContractEvent]) – List of events to sort

Returns:

Sorted list of events

Return type:

list[ContractEvent]

Example

>>> sorted_events = sort_events([event3, event1, event2])

References

ACTUS Technical Specification v1.1, Section 2.9

jactus.core.events.merge_congruent_events(event1, event2)[source]

Merge two events at the same time.

Used for composite contracts where multiple events occur simultaneously. Payoffs are summed, and the earliest event type takes precedence.

Parameters:
Returns:

Merged event

Raises:

ValueError – If events have different times or currencies

Return type:

ContractEvent

Example

>>> merged = merge_congruent_events(event1, event2)

References

ACTUS Technical Specification v1.1, Section 2.9

Contract Types

Type definitions and enumerations for ACTUS contract standard.

This module defines all enumerations and type aliases used throughout JACTUS. All enumerations inherit from str for JSON serializability and easy comparison.

References

ACTUS Technical Specification v1.1, Section 2 (Notations)

class jactus.core.types.EventType(*values)[source]

Bases: str, Enum

ACTUS contract event types.

Event types define the nature of contract events that trigger state transitions and cash flows.

References

ACTUS Technical Specification v1.1, Table 4

AD = 'AD'
IED = 'IED'
MD = 'MD'
PR = 'PR'
PI = 'PI'
PP = 'PP'
PY = 'PY'
PRF = 'PRF'
FP = 'FP'
PRD = 'PRD'
TD = 'TD'
IP = 'IP'
IPCI = 'IPCI'
IPCB = 'IPCB'
RR = 'RR'
RRF = 'RRF'
DV = 'DV'
DVF = 'DVF'
SC = 'SC'
STD = 'STD'
XD = 'XD'
CE = 'CE'
IPFX = 'IPFX'
IPFL = 'IPFL'
property index: int

Integer index for this event type (for dispatch tables and future jax.lax.switch).

class jactus.core.types.ContractType(*values)[source]

Bases: str, Enum

ACTUS contract types.

Defines all standardized contract types in the ACTUS taxonomy.

References

ACTUS Technical Specification v1.1, Table 3

PAM = 'PAM'
LAM = 'LAM'
LAX = 'LAX'
NAM = 'NAM'
ANN = 'ANN'
CLM = 'CLM'
UMP = 'UMP'
CSH = 'CSH'
STK = 'STK'
COM = 'COM'
FXOUT = 'FXOUT'
SWPPV = 'SWPPV'
SWAPS = 'SWAPS'
CAPFL = 'CAPFL'
OPTNS = 'OPTNS'
FUTUR = 'FUTUR'
CEG = 'CEG'
CEC = 'CEC'
class jactus.core.types.ContractRole(*values)[source]

Bases: str, Enum

Contract party role definition.

Defines the role of the contract creator, which determines the sign of cash flows (+1 for asset positions, -1 for liability).

References

ACTUS Technical Specification v1.1, Table 1

RPA = 'RPA'
RPL = 'RPL'
LG = 'LG'
ST = 'ST'
BUY = 'BUY'
SEL = 'SEL'
RFL = 'RFL'
PFL = 'PFL'
COL = 'COL'
CNO = 'CNO'
GUA = 'GUA'
OBL = 'OBL'
UDL = 'UDL'
UDLP = 'UDLP'
UDLM = 'UDLM'
get_sign()[source]

Get the sign convention for this role.

Returns +1 for asset/long/receive positions, -1 for liability/short/pay positions.

Returns:

+1 or -1 according to ACTUS Table 1

Return type:

int

Example

>>> ContractRole.RPA.get_sign()
1
>>> ContractRole.RPL.get_sign()
-1

References

ACTUS Technical Specification v1.1, Table 1, Section 3.7

class jactus.core.types.DayCountConvention(*values)[source]

Bases: str, Enum

Day count conventions for year fraction calculation.

Determines how time periods are calculated for interest accrual.

References

ACTUS Technical Specification v1.1, Section 3.6 ISDA Definitions

AA = 'AA'
A360 = 'A360'
A365 = 'A365'
E30360ISDA = '30E360ISDA'
E30360 = '30E360'
B30360 = '30360'
BUS252 = 'BUS252'
class jactus.core.types.BusinessDayConvention(*values)[source]

Bases: str, Enum

Business day adjustment conventions.

Defines how dates are adjusted when they fall on non-business days. Convention format: S/C + Direction + Modified - S = Shift, C = Calculate - F = Following, P = Preceding - M prefix = Modified (don’t cross month boundary)

References

ACTUS Technical Specification v1.1, Section 3.4

NULL = 'NULL'
SCF = 'SCF'
SCMF = 'SCMF'
CSF = 'CSF'
CSMF = 'CSMF'
SCP = 'SCP'
SCMP = 'SCMP'
CSP = 'CSP'
CSMP = 'CSMP'
class jactus.core.types.EndOfMonthConvention(*values)[source]

Bases: str, Enum

End of month adjustment convention.

Determines whether dates stay at month-end or maintain day number.

References

ACTUS Technical Specification v1.1, Section 3.3

EOM = 'EOM'
SD = 'SD'
class jactus.core.types.Calendar(*values)[source]

Bases: str, Enum

Business day calendar definitions.

Defines which days are considered business days (non-holidays).

Note

Additional calendars (TARGET, NYSE, etc.) can be added as needed.

NO_CALENDAR = 'NO_CALENDAR'
MONDAY_TO_FRIDAY = 'MONDAY_TO_FRIDAY'
TARGET = 'TARGET'
US_NYSE = 'US_NYSE'
UK_SETTLEMENT = 'UK_SETTLEMENT'
CUSTOM = 'CUSTOM'
class jactus.core.types.ContractPerformance(*values)[source]

Bases: str, Enum

Contract performance status.

Indicates the payment performance status of the contract.

References

ACTUS Technical Specification v1.1

PF = 'PF'
DL = 'DL'
DQ = 'DQ'
DF = 'DF'
class jactus.core.types.FeeBasis(*values)[source]

Bases: str, Enum

Fee calculation basis.

Determines whether fees are absolute amounts or percentages of notional.

A = 'A'
N = 'N'
class jactus.core.types.InterestCalculationBase(*values)[source]

Bases: str, Enum

Base for interest calculation.

Defines which notional value to use for interest calculations.

References

ACTUS Technical Specification v1.1

NT = 'NT'
NTIED = 'NTIED'
NTL = 'NTL'
class jactus.core.types.CyclePointOfInterestPayment(*values)[source]

Bases: str, Enum

When IP is calculated/paid in cycle.

Determines whether interest is paid at the beginning or end of each period.

B = 'B'
E = 'E'
class jactus.core.types.PrepaymentEffect(*values)[source]

Bases: str, Enum

Effect of prepayment on schedule.

Determines how unscheduled principal payments affect the contract.

N = 'N'
A = 'A'
M = 'M'
class jactus.core.types.ScalingEffect(*values)[source]

Bases: str, Enum

Scaling index effect on contract.

Three-character code indicating which contract elements are scaled: - Position 1: Interest (I = scale, 0 = no scale) - Position 2: Notional (N = scale, 0 = no scale) - Position 3: Maturity (M = scale, 0 = no scale)

Example

‘IN0’ = Scale Interest and Notional, but not Maturity

References

ACTUS Technical Specification v1.1

S000 = '000'
I00 = 'I00'
S0N0 = '0N0'
IN0 = 'IN0'
S00M = '00M'
I0M = 'I0M'
S0NM = '0NM'
INM = 'INM'

Utilities

Schedule Generation

Schedule generation utilities for ACTUS contracts.

This module provides functions for generating event schedules according to ACTUS specifications, including handling of cycle notation, end-of-month conventions, and business day conventions.

References

ACTUS Technical Specification v1.1, Section 3 (Schedule Generation)

jactus.utilities.schedules.generate_schedule(start, cycle, end, end_of_month_convention=EndOfMonthConvention.SD, business_day_convention=BusinessDayConvention.NULL, calendar=Calendar.NO_CALENDAR)[source]

Generate a regular event schedule S(s, c, T).

Generates dates starting from ‘start’, adding ‘cycle’ repeatedly until reaching or exceeding ‘end’. Applies end-of-month and business day conventions.

Parameters:

  • start (ActusDateTime | None) – Schedule start date (anchor)

  • cycle (str | None) – Cycle string in NPS format (e.g., ‘3M’, ‘1Y’, ‘1Q+’)

  • end (ActusDateTime | None) – Schedule end date

  • end_of_month_convention (EndOfMonthConvention) – How to handle month-end dates

  • business_day_convention (BusinessDayConvention) – How to adjust non-business days

  • calendar (Calendar) – Business day calendar to use

Returns:

List of dates in chronological order

Return type:

list[ActusDateTime]

Example

>>> schedule = generate_schedule(
...     start=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     cycle="3M",
...     end=ActusDateTime(2025, 1, 15, 0, 0, 0),
... )

References

ACTUS Technical Specification v1.1, Section 3.1

jactus.utilities.schedules.generate_array_schedule(anchors, cycles, end, end_of_month_convention=EndOfMonthConvention.SD, business_day_convention=BusinessDayConvention.NULL, calendar=Calendar.NO_CALENDAR)[source]

Generate array schedule S~(~s, ~c, T).

Generates a schedule from multiple (anchor, cycle) pairs. Each pair generates a sub-schedule that ends at the next anchor or final end.

Parameters:
Returns:

Sorted list of all dates from all sub-schedules

Return type:

list[ActusDateTime]

Example

>>> schedule = generate_array_schedule(
...     anchors=[
...         ActusDateTime(2024, 1, 15, 0, 0, 0),
...         ActusDateTime(2024, 7, 15, 0, 0, 0),
...     ],
...     cycles=["3M", "6M"],
...     end=ActusDateTime(2025, 1, 15, 0, 0, 0),
... )

References

ACTUS Technical Specification v1.1, Section 3.2

jactus.utilities.schedules.apply_end_of_month_convention(dates, start, cycle, convention)[source]

Apply end-of-month convention to schedule.

The EOM convention only applies if: 1. Start date is the last day of a month with <31 days 2. Cycle is a multiple of 1 month

Parameters:
Returns:

List of adjusted dates

Return type:

list[ActusDateTime]

References

ACTUS Technical Specification v1.1, Section 3.3

jactus.utilities.schedules.apply_business_day_convention(dates, convention, calendar)[source]

Apply business day convention to schedule.

Adjusts each date according to the specified convention if it falls on a non-business day.

Parameters:
Returns:

List of adjusted dates

Return type:

list[ActusDateTime]

References

ACTUS Technical Specification v1.1, Section 3.4

jactus.utilities.schedules.expand_period_to_months(period, multiplier)[source]

Convert period to number of months.

Parameters:

  • period (str) – Period type (D, W, M, Q, H, Y)

  • multiplier (int) – Number of periods

Returns:

Number of months, or None for D/W (day/week periods)

Return type:

int | None

Example

>>> expand_period_to_months('Q', 2)
6
>>> expand_period_to_months('Y', 1)
12

Business Day Conventions

Day count convention implementations for ACTUS contracts.

This module provides year fraction calculations according to various day count conventions used in financial contracts.

References

ACTUS Technical Specification v1.1, Section 4 (Day Count Conventions) ISDA 2006 Definitions

jactus.utilities.conventions.year_fraction(start, end, convention, maturity=None, calendar=None)[source]

Calculate year fraction between two dates using specified convention.

Parameters:

  • start (ActusDateTime) – Start date

  • end (ActusDateTime) – End date

  • convention (DayCountConvention) – Day count convention to use

  • maturity (ActusDateTime | None) – Maturity date (required for some conventions)

  • calendar (HolidayCalendar | None) – Holiday calendar for BUS/252 convention. If None, defaults to MondayToFridayCalendar (Mon-Fri only, no public holidays).

Returns:

Year fraction as a float

Return type:

float

Example

>>> start = ActusDateTime(2024, 1, 15, 0, 0, 0)
>>> end = ActusDateTime(2024, 7, 15, 0, 0, 0)
>>> year_fraction(start, end, DayCountConvention.AA)
0.5

References

ACTUS Technical Specification v1.1, Section 4.1

jactus.utilities.conventions.days_between_30_360_methods(start, end, method='30E/360')[source]

Calculate days between two dates using 30/360 methods.

Parameters:
Returns:

Number of days (can be negative if end < start)

Return type:

int

Example

>>> start = ActusDateTime(2024, 2, 15, 0, 0, 0)
>>> end = ActusDateTime(2024, 8, 15, 0, 0, 0)
>>> days_between_30_360_methods(start, end, "30E/360")
180

Calendars

Business day calendar implementations for ACTUS contracts.

This module provides holiday calendar functionality for determining business days and adjusting dates according to business day conventions.

References

ACTUS Technical Specification v1.1, Section 3.4 (Business Day Conventions)

class jactus.utilities.calendars.HolidayCalendar[source]

Bases: ABC

Abstract base class for holiday calendars.

A holiday calendar determines which dates are business days and provides navigation functions for working with business days.

abstractmethod is_business_day(date)[source]

Check if a date is a business day.

Parameters:

date (ActusDateTime) – Date to check

Returns:

True if the date is a business day

Return type:

bool

Example

>>> cal = MondayToFridayCalendar()
>>> cal.is_business_day(ActusDateTime(2024, 1, 15, 0, 0, 0))  # Monday
True
is_holiday(date)[source]

Check if a date is a holiday (not a business day).

Parameters:

date (ActusDateTime) – Date to check

Returns:

True if the date is a holiday

Return type:

bool

next_business_day(date)[source]

Get the next business day on or after the given date.

Parameters:

date (ActusDateTime) – Starting date

Returns:

Next business day (may be the same date if already a business day)

Return type:

ActusDateTime

Example

>>> cal = MondayToFridayCalendar()
>>> saturday = ActusDateTime(2024, 1, 6, 0, 0, 0)
>>> cal.next_business_day(saturday)
ActusDateTime(2024, 1, 8, 0, 0, 0)  # Monday
previous_business_day(date)[source]

Get the previous business day on or before the given date.

Parameters:

date (ActusDateTime) – Starting date

Returns:

Previous business day (may be the same date if already a business day)

Return type:

ActusDateTime

Example

>>> cal = MondayToFridayCalendar()
>>> sunday = ActusDateTime(2024, 1, 7, 0, 0, 0)
>>> cal.previous_business_day(sunday)
ActusDateTime(2024, 1, 5, 0, 0, 0)  # Friday
add_business_days(date, days)[source]

Add a number of business days to a date.

Parameters:

  • date (ActusDateTime) – Starting date

  • days (int) – Number of business days to add (can be negative)

Returns:

Date after adding the specified business days

Return type:

ActusDateTime

Example

>>> cal = MondayToFridayCalendar()
>>> friday = ActusDateTime(2024, 1, 5, 0, 0, 0)
>>> cal.add_business_days(friday, 1)  # Next business day
ActusDateTime(2024, 1, 8, 0, 0, 0)  # Monday
business_days_between(start, end, include_end=False)[source]

Count business days between two dates.

Parameters:
Returns:

Number of business days

Return type:

int

Example

>>> cal = MondayToFridayCalendar()
>>> mon = ActusDateTime(2024, 1, 1, 0, 0, 0)
>>> fri = ActusDateTime(2024, 1, 5, 0, 0, 0)
>>> cal.business_days_between(mon, fri)
4
class jactus.utilities.calendars.NoHolidayCalendar[source]

Bases: HolidayCalendar

Calendar with no holidays - every day is a business day.

Useful for theoretical calculations or contracts that don’t respect weekends.

is_business_day(date)[source]

Every day is a business day.

Parameters:

date (ActusDateTime) – Date to check

Returns:

Always True

Return type:

bool

class jactus.utilities.calendars.MondayToFridayCalendar[source]

Bases: HolidayCalendar

Calendar with Monday-Friday as business days (no holidays).

Treats weekends (Saturday/Sunday) as non-business days but doesn’t account for public holidays.

is_business_day(date)[source]

Check if date is Monday-Friday.

Parameters:

date (ActusDateTime) – Date to check

Returns:

True if Monday-Friday, False if Saturday-Sunday

Return type:

bool

class jactus.utilities.calendars.CustomCalendar(holidays=None, include_weekends=True)[source]

Bases: HolidayCalendar

Calendar with custom holiday dates.

Allows specifying specific dates as holidays in addition to weekends.

Parameters:
__init__(holidays=None, include_weekends=True)[source]

Initialize custom calendar.

Parameters:

  • holidays (list[ActusDateTime] | None) – List of holiday dates (defaults to empty)

  • include_weekends (bool) – Whether weekends are also holidays (default True)

add_holiday(date)[source]

Add a holiday to the calendar.

Parameters:

date (ActusDateTime) – Date to mark as holiday

Return type:

None

remove_holiday(date)[source]

Remove a holiday from the calendar.

Parameters:

date (ActusDateTime) – Date to remove from holidays

Return type:

None

load_from_list(holidays)[source]

Load holidays from a list of dates.

Parameters:

holidays (list[ActusDateTime]) – List of holiday dates

Return type:

None

is_business_day(date)[source]

Check if date is a business day.

A business day is not a weekend (if include_weekends=True) and not in the custom holidays list.

Parameters:

date (ActusDateTime) – Date to check

Returns:

True if business day, False if weekend or holiday

Return type:

bool

class jactus.utilities.calendars.TARGETCalendar[source]

Bases: HolidayCalendar

ECB TARGET2 calendar.

TARGET (Trans-European Automated Real-time Gross Settlement Express Transfer) holidays: New Year’s Day, Good Friday, Easter Monday, Labour Day (May 1), Christmas Day, Boxing Day.

Holiday dates are pre-computed for years 2000-2100.

__init__()[source]
Return type:

None

is_business_day(dt)[source]

Check if date is a TARGET business day.

Parameters:

dt (ActusDateTime)

Return type:

bool

class jactus.utilities.calendars.NYSECalendar[source]

Bases: HolidayCalendar

New York Stock Exchange calendar.

NYSE holidays: New Year’s Day, MLK Day (3rd Mon Jan), Presidents’ Day (3rd Mon Feb), Good Friday, Memorial Day (last Mon May), Juneteenth (Jun 19), Independence Day (Jul 4), Labor Day (1st Mon Sep), Thanksgiving (4th Thu Nov), Christmas Day.

Holiday dates are pre-computed for years 2000-2100.

__init__()[source]
Return type:

None

is_business_day(dt)[source]

Check if date is an NYSE business day.

Parameters:

dt (ActusDateTime)

Return type:

bool

class jactus.utilities.calendars.UKSettlementCalendar[source]

Bases: HolidayCalendar

UK Settlement (bank holidays) calendar.

UK bank holidays: New Year’s Day, Good Friday, Easter Monday, Early May Bank Holiday (1st Mon May), Spring Bank Holiday (last Mon May), Summer Bank Holiday (last Mon Aug), Christmas Day, Boxing Day.

Holiday dates are pre-computed for years 2000-2100.

__init__()[source]
Return type:

None

is_business_day(dt)[source]

Check if date is a UK Settlement business day.

Parameters:

dt (ActusDateTime)

Return type:

bool

jactus.utilities.calendars.get_calendar(calendar_name)[source]

Factory function to get a calendar by name.

Parameters:

calendar_name (str) – Name of calendar (“NO_CALENDAR”, “MONDAY_TO_FRIDAY”, etc.)

Returns:

HolidayCalendar instance

Raises:

ValueError – If calendar name is unknown

Return type:

HolidayCalendar

Example

>>> cal = get_calendar("MONDAY_TO_FRIDAY")
>>> cal.is_business_day(ActusDateTime(2024, 1, 6, 0, 0, 0))  # Saturday
False
jactus.utilities.calendars.is_weekend(date)[source]

Check if a date falls on a weekend (Saturday or Sunday).

Parameters:

date (ActusDateTime) – Date to check

Returns:

True if Saturday or Sunday

Return type:

bool

Example

>>> is_weekend(ActusDateTime(2024, 1, 6, 0, 0, 0))  # Saturday
True
>>> is_weekend(ActusDateTime(2024, 1, 8, 0, 0, 0))  # Monday
False

Mathematical Functions

Financial mathematics utilities for ACTUS contracts.

This module provides mathematical functions for contract calculations including contract role signs, annuity calculations, and discount factors.

References

ACTUS Technical Specification v1.1, Table 1 (Contract Role Signs) ACTUS Technical Specification v1.1, Section 5 (Mathematical Functions)

jactus.utilities.math.contract_role_sign(role)[source]

Get the sign (+1 or -1) for a contract role.

The contract role sign determines the direction of cash flows from the perspective of the contract holder.

Parameters:

role (ContractRole) – Contract role

Returns:

+1 for long/receiving positions, -1 for short/paying positions

Return type:

int

Example

>>> contract_role_sign(ContractRole.RPA)  # Receiving party A
1
>>> contract_role_sign(ContractRole.RPL)  # Real position lender
-1

References

ACTUS Technical Specification v1.1, Table 1

jactus.utilities.math.contract_role_sign_vectorized(roles)[source]

Vectorized contract role sign calculation for JAX.

Parameters:

roles (Array) – Array of contract role values (as integers)

Returns:

Array of signs (+1 or -1)

Return type:

Array

Example

>>> roles = jnp.array([0, 1, 2])  # RPA, RPL, LG
>>> signs = contract_role_sign_vectorized(roles)
>>> signs
Array([1, -1, 1], dtype=int32)

Note

This function is JIT-compiled for performance.

jactus.utilities.math.annuity_amount(notional, rate, tenor, maturity, n_periods, day_count_convention)[source]

Calculate annuity payment amount.

Computes the periodic payment for an annuity given notional, rate, and term. Uses the formula: A = N * r / (1 - (1 + r)^(-n))

Parameters:

  • notional (float) – Notional principal amount

  • rate (float) – Periodic interest rate (e.g., 0.05 for 5% per period)

  • tenor (ActusDateTime) – Start date for year fraction calculation (reserved for future use)

  • maturity (ActusDateTime) – End date for year fraction calculation (reserved for future use)

  • n_periods (int) – Number of payment periods

  • day_count_convention (DayCountConvention) – Day count convention (reserved for future use)

Returns:

Annuity payment amount per period

Return type:

float

Example

>>> # $100,000 loan at 5% annual rate, 12 monthly payments
>>> tenor = ActusDateTime(2024, 1, 1, 0, 0, 0)
>>> maturity = ActusDateTime(2025, 1, 1, 0, 0, 0)
>>> amount = annuity_amount(100000, 0.05/12, tenor, maturity, 12, DayCountConvention.A360)
>>> abs(amount - 8560.75) < 1  # Approximately $8,560.75 per month
True

References

ACTUS Technical Specification v1.1, Section 5.1

jactus.utilities.math.annuity_amount_vectorized(notional, rate, n_periods)[source]

Vectorized annuity calculation for JAX arrays.

Parameters:

  • notional (Array) – Array of notional amounts

  • rate (Array) – Array of periodic rates

  • n_periods (Array) – Array of number of periods

Returns:

Array of annuity amounts

Return type:

Array

Example

>>> notionals = jnp.array([100000.0, 200000.0])
>>> rates = jnp.array([0.05/12, 0.04/12])
>>> periods = jnp.array([12, 24])
>>> amounts = annuity_amount_vectorized(notionals, rates, periods)

Note

This function is JIT-compiled for performance.

jactus.utilities.math.discount_factor(rate, start, end, day_count_convention)[source]

Calculate discount factor for a time period.

Computes: DF = 1 / (1 + r * t) where t is the year fraction between start and end.

Parameters:
Returns:

Discount factor

Return type:

float

Example

>>> start = ActusDateTime(2024, 1, 1, 0, 0, 0)
>>> end = ActusDateTime(2024, 7, 1, 0, 0, 0)
>>> df = discount_factor(0.05, start, end, DayCountConvention.AA)
>>> abs(df - 0.9756) < 0.001  # Approximately 0.9756
True

References

ACTUS Technical Specification v1.1, Section 5.2

jactus.utilities.math.discount_factor_vectorized(rate, year_fraction)[source]

Vectorized discount factor calculation.

Parameters:

  • rate (Array) – Array of interest rates

  • year_fraction (Array) – Array of year fractions

Returns:

Array of discount factors

Return type:

Array

Example

>>> rates = jnp.array([0.05, 0.04, 0.06])
>>> yfs = jnp.array([0.5, 1.0, 0.25])
>>> dfs = discount_factor_vectorized(rates, yfs)

Note

This function is JIT-compiled for performance.

jactus.utilities.math.compound_factor(rate, start, end, day_count_convention, compounding_frequency=1)[source]

Calculate compound factor for a time period.

Computes: CF = (1 + r/m)^(m*t) where m is the compounding frequency and t is the year fraction.

Parameters:

  • rate (float) – Annual interest rate (e.g., 0.05 for 5%)

  • start (ActusDateTime) – Start date

  • end (ActusDateTime) – End date

  • day_count_convention (DayCountConvention) – Day count convention

  • compounding_frequency (int) – Number of compounding periods per year (default 1)

Returns:

Compound factor

Return type:

float

Example

>>> start = ActusDateTime(2024, 1, 1, 0, 0, 0)
>>> end = ActusDateTime(2025, 1, 1, 0, 0, 0)
>>> # Annual compounding
>>> cf = compound_factor(0.05, start, end, DayCountConvention.AA, 1)
>>> abs(cf - 1.05) < 0.001
True
>>> # Monthly compounding
>>> cf_monthly = compound_factor(0.05, start, end, DayCountConvention.AA, 12)
>>> abs(cf_monthly - 1.05116) < 0.001
True

References

Standard financial mathematics

jactus.utilities.math.compound_factor_vectorized(rate, year_fraction, compounding_frequency)[source]

Vectorized compound factor calculation.

Parameters:

  • rate (Array) – Array of interest rates

  • year_fraction (Array) – Array of year fractions

  • compounding_frequency (Array) – Array of compounding frequencies

Returns:

Array of compound factors

Return type:

Array

Example

>>> rates = jnp.array([0.05, 0.04])
>>> yfs = jnp.array([1.0, 1.0])
>>> freqs = jnp.array([1, 12])
>>> cfs = compound_factor_vectorized(rates, yfs, freqs)

Note

This function is JIT-compiled for performance. For continuous compounding (frequency=0), use a very large frequency instead.

jactus.utilities.math.present_value(cash_flows, dates, valuation_date, discount_rate, day_count_convention)[source]

Calculate present value of a series of cash flows.

Parameters:
Returns:

Present value of all cash flows

Return type:

float

Example

>>> cfs = [100, 100, 100]
>>> dates = [
...     ActusDateTime(2024, 1, 1, 0, 0, 0),
...     ActusDateTime(2024, 7, 1, 0, 0, 0),
...     ActusDateTime(2025, 1, 1, 0, 0, 0),
... ]
>>> val_date = ActusDateTime(2024, 1, 1, 0, 0, 0)
>>> pv = present_value(cfs, dates, val_date, 0.05, DayCountConvention.AA)
>>> abs(pv - 295.14) < 1.0
True

References

Standard financial mathematics

jactus.utilities.math.present_value_vectorized(cash_flows, year_fractions, discount_rate)[source]

Vectorized present value calculation.

Parameters:

  • cash_flows (Array) – Array of cash flow amounts

  • year_fractions (Array) – Array of year fractions from valuation date

  • discount_rate (float) – Discount rate

Returns:

Present value (scalar JAX array)

Return type:

Array

Example

>>> cfs = jnp.array([100.0, 100.0, 100.0])
>>> yfs = jnp.array([0.0, 0.5, 1.0])
>>> pv = present_value_vectorized(cfs, yfs, 0.05)

Note

This function is JIT-compiled for performance. Assumes all cash flows are in the future (year_fractions >= 0).

jactus.utilities.math.calculate_actus_annuity(start, pr_schedule, notional, accrued_interest, rate, day_count_convention)[source]

Calculate annuity amount using ACTUS specification formula.

Implements the ACTUS annuity formula from Section 3.8:

A(s, T, n, a, r) = (n + a) / Σ[∏((1 + Y_i × r)^-1)]

Where:

s = start time T = maturity (last PR date) n = notional principal a = accrued interest r = nominal interest rate Y_i = year fraction for period i Σ = sum over all PR events ∏ = product up to each PR event

This calculates the constant payment amount such that the total of all payments exactly amortizes the notional plus accrued interest.

Parameters:

  • start (ActusDateTime) – Start time for calculation

  • pr_schedule (list[ActusDateTime]) – List of principal redemption dates

  • notional (float) – Notional principal amount

  • accrued_interest (float) – Already accrued interest

  • rate (float) – Annual interest rate (e.g., 0.05 for 5%)

  • day_count_convention (DayCountConvention) – Day count convention for year fractions

Returns:

Annuity payment amount per period

Return type:

float

Example

>>> # $100,000 loan at 5% for 12 months
>>> start = ActusDateTime(2024, 1, 15, 0, 0, 0)
>>> pr_dates = [ActusDateTime(2024, i, 15, 0, 0, 0) for i in range(2, 14)]
>>> amount = calculate_actus_annuity(
...     start, pr_dates, 100000.0, 0.0, 0.05, DayCountConvention.A360
... )
>>> 8500 < amount < 8600  # Approximately $8,560
True

References

ACTUS Technical Specification v1.1, Section 3.8

jactus.utilities.math.calculate_actus_annuity_jax(year_fractions, notional, accrued_interest, rate)[source]

JAX-compiled version of ACTUS annuity calculation.

Implements the ACTUS annuity formula:

A(s, T, n, a, r) = (n + a) / Σ[∏((1 + Y_i × r)^-1)]

Parameters:

  • year_fractions (Array) – Array of year fractions for each period

  • notional (float) – Notional principal amount

  • accrued_interest (float) – Already accrued interest

  • rate (float) – Annual interest rate

Returns:

Annuity payment amount

Return type:

float

Example

>>> # 12 equal monthly periods (30/360 convention)
>>> yfs = jnp.array([30/360] * 12)
>>> amount = calculate_actus_annuity_jax(yfs, 100000.0, 0.0, 0.05)

Note

This function is JIT-compiled for performance.

References

ACTUS Technical Specification v1.1, Section 3.8

2D Surface Interpolation

2D surface interpolation for behavioral risk models.

This module provides JAX-compatible 2D surface interpolation, used primarily by behavioral risk models such as prepayment and deposit transaction models.

A Surface2D represents a function f(x, y) defined on a grid of (x, y) margin values with corresponding z values. Given arbitrary query points (x_q, y_q), it returns interpolated values using bilinear interpolation.

Key features: - Bilinear interpolation on 2D grids - Configurable extrapolation: "constant" (nearest edge) or "raise" - JAX array storage for automatic differentiation compatibility - Support for both numeric and label-based margins via LabeledSurface2D

References

ACTUS Risk Service v2.0 - TimeSeries<TimeSeries<Double>> surface model

class jactus.utilities.surface.Surface2D(x_margins, y_margins, values, extrapolation='constant')[source]

Bases: object

A 2D interpolation surface defined on a rectangular grid.

The surface is specified by: - x_margins: Sorted 1D array of x-axis breakpoints (e.g., spread values) - y_margins: Sorted 1D array of y-axis breakpoints (e.g., age values) - values: 2D array of shape (len(x_margins), len(y_margins))

Interpolation is bilinear within the grid, and extrapolation behavior is configurable.

Parameters:
x_margins

Sorted 1D array of x-axis breakpoints.

Type:

jax.Array

y_margins

Sorted 1D array of y-axis breakpoints.

Type:

jax.Array

values

2D array of shape (len(x_margins), len(y_margins)).

Type:

jax.Array

extrapolation

Extrapolation method: "constant" (use nearest edge value) or "raise" (raise ValueError).

Type:

str

Example

>>> import jax.numpy as jnp
>>> # Prepayment surface: spread (rows) × age (columns)
>>> surface = Surface2D(
...     x_margins=jnp.array([0.0, 1.0, 2.0, 3.0]),   # spread %
...     y_margins=jnp.array([0.0, 1.0, 3.0, 5.0]),    # age years
...     values=jnp.array([
...         [0.00, 0.00, 0.00, 0.00],  # spread=0%
...         [0.00, 0.01, 0.02, 0.00],  # spread=1%
...         [0.00, 0.02, 0.05, 0.01],  # spread=2%
...         [0.01, 0.05, 0.10, 0.02],  # spread=3%
...     ]),
... )
>>> rate = surface.evaluate(1.5, 2.0)  # Bilinear interpolation
x_margins: Array
y_margins: Array
values: Array
extrapolation: str = 'constant'
__post_init__()[source]

Validate surface dimensions and parameters.

Return type:

None

evaluate(x, y)[source]

Evaluate the surface at a point using bilinear interpolation.

For points within the grid, standard bilinear interpolation is used. For points outside the grid, behavior depends on self.extrapolation.

Parameters:

  • x (float) – X-axis query value.

  • y (float) – Y-axis query value.

Returns:

Interpolated value as a scalar JAX array.

Raises:

ValueError – If extrapolation="raise" and point is outside the grid.

Return type:

Array

Example

>>> value = surface.evaluate(1.5, 2.0)
static from_dict(data)[source]

Create a Surface2D from a dictionary representation.

Parameters:

data (dict[str, Any]) – Dictionary with keys "x_margins", "y_margins", "values", and optionally "extrapolation".

Returns:

New Surface2D instance.

Return type:

Surface2D

Example

>>> surface = Surface2D.from_dict({
...     "x_margins": [0.0, 1.0, 2.0],
...     "y_margins": [0.0, 1.0, 3.0],
...     "values": [
...         [0.0, 0.01, 0.02],
...         [0.01, 0.03, 0.05],
...         [0.02, 0.05, 0.10],
...     ],
...     "extrapolation": "constant",
... })
to_dict()[source]

Serialize surface to a dictionary.

Returns:

Dictionary representation suitable for JSON serialization.

Return type:

dict[str, Any]

__init__(x_margins, y_margins, values, extrapolation='constant')
Parameters:
Return type:

None

class jactus.utilities.surface.LabeledSurface2D(x_labels, y_labels, values)[source]

Bases: object

A 2D surface with string-labeled margins.

Used for models where one or both dimensions are categorical rather than numeric (e.g., deposit transaction models where dimension 1 is contract ID and dimension 2 is a date label).

Each label maps to a numeric index. The underlying data is stored in a regular Surface2D for interpolation.

Parameters:
x_labels

Ordered list of labels for the x-axis.

Type:

list[str]

y_labels

Ordered list of labels for the y-axis.

Type:

list[str]

values

2D array of shape (len(x_labels), len(y_labels)).

Type:

jax.Array

Example

>>> surface = LabeledSurface2D(
...     x_labels=["DEPOSIT-001", "DEPOSIT-002"],
...     y_labels=["2024-01-01", "2024-07-01", "2025-01-01"],
...     values=jnp.array([
...         [1000.0, -500.0, 2000.0],
...         [5000.0, -1000.0, 3000.0],
...     ]),
... )
>>> amount = surface.get(x_label="DEPOSIT-001", y_label="2024-07-01")
x_labels: list[str]
y_labels: list[str]
values: Array
__post_init__()[source]

Validate dimensions.

Return type:

None

get(x_label, y_label)[source]

Get the value at exact label coordinates.

Parameters:

  • x_label (str) – X-axis label.

  • y_label (str) – Y-axis label.

Returns:

Value as a scalar JAX array.

Raises:

KeyError – If either label is not found.

Return type:

Array

Example

>>> value = surface.get("DEPOSIT-001", "2024-07-01")
get_row(x_label)[source]

Get all values for a given x-axis label.

Parameters:

x_label (str) – X-axis label.

Returns:

1D array of values for that row.

Raises:

KeyError – If label not found.

Return type:

Array

get_column(y_label)[source]

Get all values for a given y-axis label.

Parameters:

y_label (str) – Y-axis label.

Returns:

1D array of values for that column.

Raises:

KeyError – If label not found.

Return type:

Array

static from_dict(data)[source]

Create from dictionary representation.

Parameters:

data (dict[str, Any]) – Dictionary with keys "x_labels", "y_labels", "values".

Returns:

New LabeledSurface2D instance.

Return type:

LabeledSurface2D

to_dict()[source]

Serialize to dictionary.

Return type:

dict[str, Any]

__init__(x_labels, y_labels, values)
Parameters:
Return type:

None

Contract Implementations

Base Contract

Base contract class for all ACTUS contracts.

This module implements the abstract base class that all ACTUS contract types inherit from. It provides the core simulation engine and common functionality.

References

ACTUS v1.1 Section 3 - Contract Types ACTUS v1.1 Section 4 - Event Schedules

class jactus.contracts.base.SimulationHistory(events, states, initial_state, final_state)[source]

Bases: object

Results from contract simulation.

Contains the complete history of events and states from a contract simulation run.

Parameters:
events

List of all events (scheduled + observed)

Type:

list[jactus.core.events.ContractEvent]

states

List of states (one per event, plus initial)

Type:

list[jactus.core.states.ContractState]

initial_state

Contract state before first event. When IED >= SD (contract hasn’t started yet), this is the pre-IED state with nt=0. When IED < SD (contract already existed at status date), this reflects the post-IED state reconstructed from contract attributes (nt=notional, ipnr=rate, etc.).

Type:

jactus.core.states.ContractState

final_state

Contract state after last event

Type:

jactus.core.states.ContractState

Example

>>> history = contract.simulate(observers)
>>> print(f"Generated {len(history.events)} events")
>>> print(f"Final notional: {history.final_state.nt}")
events: list[ContractEvent]
states: list[ContractState]
initial_state: ContractState
final_state: ContractState
get_cashflows()[source]

Extract cashflow timeline from events.

Returns:

List of (time, payoff, currency) tuples

Return type:

list[tuple[ActusDateTime, Array, str]]

Example

>>> cashflows = history.get_cashflows()
>>> for time, amount, currency in cashflows:
...     print(f"{time.to_iso()}: {amount} {currency}")
filter_events(start=None, end=None)[source]

Filter events by time range.

Parameters:
Returns:

List of events in the specified range

Return type:

list[ContractEvent]

Example

>>> year_events = history.filter_events(
...     start=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     end=ActusDateTime(2024, 12, 31, 23, 59, 59)
... )
__init__(events, states, initial_state, final_state)
Parameters:
Return type:

None

class jactus.contracts.base.BaseContract(*args, **kwargs)[source]

Bases: Module, ABC

Abstract base class for all ACTUS contracts.

This class provides the core simulation engine and common functionality that all contract types share. Subclasses must implement the abstract methods to define contract-specific behavior.

The class extends flax.nnx.Module for Pytree compatibility with JAX. Note: the scalar simulation path (this class) processes events sequentially in Python and does not support JIT/grad/vmap. For JIT-compiled batch simulation with autodiff, use the array-mode API (see *_array.py modules and simulate_portfolio()).

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes (terms and conditions)

risk_factor_observer

Observer for market risk factors

child_contract_observer

Optional observer for child contracts

_event_cache

Cached event schedule (None until first computation)

Example

>>> class MyContract(BaseContract):
...     def generate_event_schedule(self):
...         # Generate contract-specific events
...         pass
...     # ... implement other abstract methods
>>> contract = MyContract(attributes, risk_observer)
>>> history = contract.simulate()
>>> cashflows = history.get_cashflows()

References

ACTUS v1.1 Section 3 - Contract Types ACTUS v1.1 Section 4 - Algorithm

__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize base contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes (terms and conditions)

  • risk_factor_observer (RiskFactorObserver) – Observer for accessing market risk factors

  • child_contract_observer (ChildContractObserver | None) – Optional observer for child contracts

  • rngs (Rngs | None) – Optional Flax RNG state for stochastic contracts

Example

>>> from jactus.observers import ConstantRiskFactorObserver
>>> attrs = ContractAttributes(...)
>>> risk_obs = ConstantRiskFactorObserver(1.0)
>>> contract = MyContract(attrs, risk_obs)
abstractmethod generate_event_schedule()[source]

Generate the scheduled events for this contract.

This method must be implemented by each contract type to generate its specific event schedule according to ACTUS rules.

Returns:

EventSchedule containing all scheduled events

Return type:

EventSchedule

Example

>>> def generate_event_schedule(self):
...     events = []
...     # Add IED event
...     events.append(ContractEvent(
...         event_type=EventType.IED,
...         event_time=self.attributes.initial_exchange_date,
...         ...
...     ))
...     # Add other events...
...     return EventSchedule(events)

References

ACTUS v1.1 Section 4.1 - Event Schedule Generation

abstractmethod initialize_state()[source]

Initialize contract state before first event.

Creates the initial state based on contract attributes. This state is used as the starting point for simulation.

Returns:

Initial ContractState

Return type:

ContractState

Example

>>> def initialize_state(self):
...     return ContractState(
...         sd=self.attributes.status_date,
...         tmd=self.attributes.maturity_date,
...         nt=jnp.array(self.attributes.notional_principal),
...         ipnr=jnp.array(self.attributes.nominal_interest_rate),
...         ...
...     )

References

ACTUS v1.1 Section 4.2 - State Initialization

abstractmethod get_payoff_function(event_type)[source]

Get payoff function for a specific event type.

Returns the appropriate payoff function (POF) for calculating the cashflow generated by the given event type.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

PayoffFunction for the event type

Return type:

PayoffFunction

Example

>>> def get_payoff_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentPayoff()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionPayoff()
...     else:
...         return ZeroPayoff()

References

ACTUS v1.1 Section 2.7 - Payoff Functions

abstractmethod get_state_transition_function(event_type)[source]

Get state transition function for a specific event type.

Returns the appropriate state transition function (STF) for updating contract state when the given event occurs.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

StateTransitionFunction for the event type

Return type:

StateTransitionFunction

Example

>>> def get_state_transition_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentSTF()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionSTF()
...     else:
...         return IdentitySTF()

References

ACTUS v1.1 Section 2.8 - State Transition Functions

get_lifetime()[source]

Get contract lifetime (start and end dates).

Returns:

Tuple of (start_date, end_date)

Return type:

tuple[ActusDateTime, ActusDateTime]

Example

>>> start, end = contract.get_lifetime()
>>> print(f"Contract runs from {start.to_iso()} to {end.to_iso()}")
is_maturity_contract()[source]

Check if contract has a defined maturity date.

Returns:

True if contract has maturity date, False otherwise

Return type:

bool

Example

>>> if contract.is_maturity_contract():
...     print("Contract matures at", contract.attributes.maturity_date)
get_events(force_regenerate=False)[source]

Get event schedule with caching.

Generates and caches the event schedule on first call. Subsequent calls return the cached schedule unless force_regenerate=True.

Parameters:

force_regenerate (bool) – If True, regenerate schedule even if cached

Returns:

EventSchedule containing all scheduled events

Return type:

EventSchedule

Example

>>> events = contract.get_events()
>>> print(f"Contract has {len(events)} events")
>>> # Regenerate if attributes changed
>>> events = contract.get_events(force_regenerate=True)
get_events_in_range(start=None, end=None)[source]

Get events within a time range.

Parameters:
Returns:

List of events in the specified range

Return type:

list[ContractEvent]

Example

>>> # Get all events in 2024
>>> events_2024 = contract.get_events_in_range(
...     start=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     end=ActusDateTime(2024, 12, 31, 23, 59, 59)
... )
simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate contract through all events.

Executes the full ACTUS algorithm:

  1. Collect callout events from behavioral observers (if any)

  2. Merge callout events into the scheduled event timeline

  3. Initialize state

  4. For each event: calculate payoff (POF) using pre-event state, apply state transition function (STF), and store event with states

Behavioral observers can be provided in three ways: - Via a Scenario object (recommended for production use) - Via the behavior_observers list parameter - By passing a BehaviorRiskFactorObserver as the risk_factor_observer

Parameters:

  • risk_factor_observer (RiskFactorObserver | None) – Optional override for risk factor observer.

  • child_contract_observer (ChildContractObserver | None) – Optional override for child contract observer.

  • scenario (Scenario | None) – Optional Scenario bundling market + behavioral observers. If provided, its market observer is used as the risk factor observer (unless risk_factor_observer is also provided), and its behavioral observers are activated for callout events.

  • behavior_observers (list[BehaviorRiskFactorObserver] | None) – Optional list of behavioral observers to activate for callout event injection.

Returns:

SimulationHistory with events and states.

Return type:

SimulationHistory

Example

>>> # Simple simulation (no behavioral models)
>>> history = contract.simulate()
>>>
>>> # With a scenario
>>> history = contract.simulate(scenario=my_scenario)
>>>
>>> # With explicit behavioral observers
>>> history = contract.simulate(
...     behavior_observers=[prepayment_model],
... )

References

ACTUS v1.1 Section 4 - Algorithm

get_cashflows(risk_factor_observer=None, child_contract_observer=None)[source]

Get cashflow timeline from contract.

Convenience method that simulates and extracts cashflows.

Parameters:

  • risk_factor_observer (RiskFactorObserver | None) – Optional override for risk factor observer

  • child_contract_observer (ChildContractObserver | None) – Optional override for child contract observer

Returns:

List of (time, payoff, currency) tuples

Return type:

list[tuple[ActusDateTime, Array, str]]

Example

>>> cashflows = contract.get_cashflows()
>>> total = sum(payoff for _, payoff, _ in cashflows)
>>> print(f"Total cashflows: {total}")
validate()[source]

Validate contract attributes.

Checks contract attributes for consistency and completeness. Returns any validation errors or warnings.

Returns:

Dictionary with ‘errors’ and ‘warnings’ lists

Return type:

dict[str, list[str]]

Example

>>> result = contract.validate()
>>> if result['errors']:
...     print("Validation failed:", result['errors'])
>>> if result['warnings']:
...     print("Warnings:", result['warnings'])

Note

Base implementation performs basic checks. Subclasses should override to add contract-specific validation.

jactus.contracts.base.sort_events_by_sequence(events)[source]

Sort events by time and sequence number.

Events are sorted first by time, then by sequence number for events at the same time. This ensures deterministic event ordering.

Parameters:

events (list[ContractEvent]) – List of events to sort

Returns:

Sorted list of events

Return type:

list[ContractEvent]

Example

>>> events = [event3, event1, event2]
>>> sorted_events = sort_events_by_sequence(events)
>>> assert sorted_events[0].event_time <= sorted_events[1].event_time

Note

This is a pure function - it does not modify the input list.

jactus.contracts.base.merge_scheduled_and_observed_events(scheduled, observed)[source]

Merge scheduled and observed events.

Combines scheduled events (from generate_event_schedule) with observed events (from child contract or risk factor observers). Removes duplicates and sorts by time and sequence.

Parameters:
Returns:

Merged and sorted list of events

Return type:

list[ContractEvent]

Example

>>> all_events = merge_scheduled_and_observed_events(
...     scheduled_events,
...     observed_events
... )

Note

If two events have the same time and type, only the first is kept. This prevents duplicate event processing.

Factory Function

Contract type implementations for various ACTUS contract types.

This module provides: - Base contract infrastructure (BaseContract, SimulationHistory) - Concrete contract implementations (CSH, PAM, STK, COM) - Contract factory pattern for dynamic instantiation - Type registration system for extensibility

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType
>>>
>>> # Create contract using factory
>>> attrs = ContractAttributes(
...     contract_id="CSH-001",
...     contract_type=ContractType.CSH,
...     notional_principal=100000.0,
...     currency="USD",
... )
>>> contract = create_contract(attrs, risk_factor_observer)
>>> result = contract.simulate()
class jactus.contracts.BaseContract(*args, **kwargs)[source]

Abstract base class for all ACTUS contracts.

This class provides the core simulation engine and common functionality that all contract types share. Subclasses must implement the abstract methods to define contract-specific behavior.

The class extends flax.nnx.Module for Pytree compatibility with JAX. Note: the scalar simulation path (this class) processes events sequentially in Python and does not support JIT/grad/vmap. For JIT-compiled batch simulation with autodiff, use the array-mode API (see *_array.py modules and simulate_portfolio()).

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes (terms and conditions)

risk_factor_observer

Observer for market risk factors

child_contract_observer

Optional observer for child contracts

_event_cache

Cached event schedule (None until first computation)

Type:

jactus.core.events.EventSchedule | None

Example

>>> class MyContract(BaseContract):
...     def generate_event_schedule(self):
...         # Generate contract-specific events
...         pass
...     # ... implement other abstract methods
>>> contract = MyContract(attributes, risk_observer)
>>> history = contract.simulate()
>>> cashflows = history.get_cashflows()

References

ACTUS v1.1 Section 3 - Contract Types ACTUS v1.1 Section 4 - Algorithm

__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize base contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes (terms and conditions)

  • risk_factor_observer (RiskFactorObserver) – Observer for accessing market risk factors

  • child_contract_observer (ChildContractObserver | None) – Optional observer for child contracts

  • rngs (Rngs | None) – Optional Flax RNG state for stochastic contracts

Example

>>> from jactus.observers import ConstantRiskFactorObserver
>>> attrs = ContractAttributes(...)
>>> risk_obs = ConstantRiskFactorObserver(1.0)
>>> contract = MyContract(attrs, risk_obs)
abstractmethod generate_event_schedule()[source]

Generate the scheduled events for this contract.

This method must be implemented by each contract type to generate its specific event schedule according to ACTUS rules.

Returns:

EventSchedule containing all scheduled events

Return type:

EventSchedule

Example

>>> def generate_event_schedule(self):
...     events = []
...     # Add IED event
...     events.append(ContractEvent(
...         event_type=EventType.IED,
...         event_time=self.attributes.initial_exchange_date,
...         ...
...     ))
...     # Add other events...
...     return EventSchedule(events)

References

ACTUS v1.1 Section 4.1 - Event Schedule Generation

abstractmethod initialize_state()[source]

Initialize contract state before first event.

Creates the initial state based on contract attributes. This state is used as the starting point for simulation.

Returns:

Initial ContractState

Return type:

ContractState

Example

>>> def initialize_state(self):
...     return ContractState(
...         sd=self.attributes.status_date,
...         tmd=self.attributes.maturity_date,
...         nt=jnp.array(self.attributes.notional_principal),
...         ipnr=jnp.array(self.attributes.nominal_interest_rate),
...         ...
...     )

References

ACTUS v1.1 Section 4.2 - State Initialization

abstractmethod get_payoff_function(event_type)[source]

Get payoff function for a specific event type.

Returns the appropriate payoff function (POF) for calculating the cashflow generated by the given event type.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

PayoffFunction for the event type

Return type:

PayoffFunction

Example

>>> def get_payoff_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentPayoff()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionPayoff()
...     else:
...         return ZeroPayoff()

References

ACTUS v1.1 Section 2.7 - Payoff Functions

abstractmethod get_state_transition_function(event_type)[source]

Get state transition function for a specific event type.

Returns the appropriate state transition function (STF) for updating contract state when the given event occurs.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

StateTransitionFunction for the event type

Return type:

StateTransitionFunction

Example

>>> def get_state_transition_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentSTF()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionSTF()
...     else:
...         return IdentitySTF()

References

ACTUS v1.1 Section 2.8 - State Transition Functions

get_lifetime()[source]

Get contract lifetime (start and end dates).

Returns:

Tuple of (start_date, end_date)

Return type:

tuple[ActusDateTime, ActusDateTime]

Example

>>> start, end = contract.get_lifetime()
>>> print(f"Contract runs from {start.to_iso()} to {end.to_iso()}")
is_maturity_contract()[source]

Check if contract has a defined maturity date.

Returns:

True if contract has maturity date, False otherwise

Return type:

bool

Example

>>> if contract.is_maturity_contract():
...     print("Contract matures at", contract.attributes.maturity_date)
get_events(force_regenerate=False)[source]

Get event schedule with caching.

Generates and caches the event schedule on first call. Subsequent calls return the cached schedule unless force_regenerate=True.

Parameters:

force_regenerate (bool) – If True, regenerate schedule even if cached

Returns:

EventSchedule containing all scheduled events

Return type:

EventSchedule

Example

>>> events = contract.get_events()
>>> print(f"Contract has {len(events)} events")
>>> # Regenerate if attributes changed
>>> events = contract.get_events(force_regenerate=True)
get_events_in_range(start=None, end=None)[source]

Get events within a time range.

Parameters:
Returns:

List of events in the specified range

Return type:

list[ContractEvent]

Example

>>> # Get all events in 2024
>>> events_2024 = contract.get_events_in_range(
...     start=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     end=ActusDateTime(2024, 12, 31, 23, 59, 59)
... )
simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate contract through all events.

Executes the full ACTUS algorithm:

  1. Collect callout events from behavioral observers (if any)

  2. Merge callout events into the scheduled event timeline

  3. Initialize state

  4. For each event: calculate payoff (POF) using pre-event state, apply state transition function (STF), and store event with states

Behavioral observers can be provided in three ways: - Via a Scenario object (recommended for production use) - Via the behavior_observers list parameter - By passing a BehaviorRiskFactorObserver as the risk_factor_observer

Parameters:

  • risk_factor_observer (RiskFactorObserver | None) – Optional override for risk factor observer.

  • child_contract_observer (ChildContractObserver | None) – Optional override for child contract observer.

  • scenario (Scenario | None) – Optional Scenario bundling market + behavioral observers. If provided, its market observer is used as the risk factor observer (unless risk_factor_observer is also provided), and its behavioral observers are activated for callout events.

  • behavior_observers (list[BehaviorRiskFactorObserver] | None) – Optional list of behavioral observers to activate for callout event injection.

Returns:

SimulationHistory with events and states.

Return type:

SimulationHistory

Example

>>> # Simple simulation (no behavioral models)
>>> history = contract.simulate()
>>>
>>> # With a scenario
>>> history = contract.simulate(scenario=my_scenario)
>>>
>>> # With explicit behavioral observers
>>> history = contract.simulate(
...     behavior_observers=[prepayment_model],
... )

References

ACTUS v1.1 Section 4 - Algorithm

get_cashflows(risk_factor_observer=None, child_contract_observer=None)[source]

Get cashflow timeline from contract.

Convenience method that simulates and extracts cashflows.

Parameters:

  • risk_factor_observer (RiskFactorObserver | None) – Optional override for risk factor observer

  • child_contract_observer (ChildContractObserver | None) – Optional override for child contract observer

Returns:

List of (time, payoff, currency) tuples

Return type:

list[tuple[ActusDateTime, Array, str]]

Example

>>> cashflows = contract.get_cashflows()
>>> total = sum(payoff for _, payoff, _ in cashflows)
>>> print(f"Total cashflows: {total}")
validate()[source]

Validate contract attributes.

Checks contract attributes for consistency and completeness. Returns any validation errors or warnings.

Returns:

Dictionary with ‘errors’ and ‘warnings’ lists

Return type:

dict[str, list[str]]

Example

>>> result = contract.validate()
>>> if result['errors']:
...     print("Validation failed:", result['errors'])
>>> if result['warnings']:
...     print("Warnings:", result['warnings'])

Note

Base implementation performs basic checks. Subclasses should override to add contract-specific validation.

class jactus.contracts.SimulationHistory(events, states, initial_state, final_state)[source]

Results from contract simulation.

Contains the complete history of events and states from a contract simulation run.

Parameters:
events

List of all events (scheduled + observed)

Type:

list[jactus.core.events.ContractEvent]

states

List of states (one per event, plus initial)

Type:

list[jactus.core.states.ContractState]

initial_state

Contract state before first event. When IED >= SD (contract hasn’t started yet), this is the pre-IED state with nt=0. When IED < SD (contract already existed at status date), this reflects the post-IED state reconstructed from contract attributes (nt=notional, ipnr=rate, etc.).

Type:

jactus.core.states.ContractState

final_state

Contract state after last event

Type:

jactus.core.states.ContractState

Example

>>> history = contract.simulate(observers)
>>> print(f"Generated {len(history.events)} events")
>>> print(f"Final notional: {history.final_state.nt}")
events: list[ContractEvent]
states: list[ContractState]
initial_state: ContractState
final_state: ContractState
get_cashflows()[source]

Extract cashflow timeline from events.

Returns:

List of (time, payoff, currency) tuples

Return type:

list[tuple[ActusDateTime, Array, str]]

Example

>>> cashflows = history.get_cashflows()
>>> for time, amount, currency in cashflows:
...     print(f"{time.to_iso()}: {amount} {currency}")
filter_events(start=None, end=None)[source]

Filter events by time range.

Parameters:
Returns:

List of events in the specified range

Return type:

list[ContractEvent]

Example

>>> year_events = history.filter_events(
...     start=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     end=ActusDateTime(2024, 12, 31, 23, 59, 59)
... )
__init__(events, states, initial_state, final_state)
Parameters:
Return type:

None

jactus.contracts.sort_events_by_sequence(events)[source]

Sort events by time and sequence number.

Events are sorted first by time, then by sequence number for events at the same time. This ensures deterministic event ordering.

Parameters:

events (list[ContractEvent]) – List of events to sort

Returns:

Sorted list of events

Return type:

list[ContractEvent]

Example

>>> events = [event3, event1, event2]
>>> sorted_events = sort_events_by_sequence(events)
>>> assert sorted_events[0].event_time <= sorted_events[1].event_time

Note

This is a pure function - it does not modify the input list.

jactus.contracts.merge_scheduled_and_observed_events(scheduled, observed)[source]

Merge scheduled and observed events.

Combines scheduled events (from generate_event_schedule) with observed events (from child contract or risk factor observers). Removes duplicates and sorts by time and sequence.

Parameters:
Returns:

Merged and sorted list of events

Return type:

list[ContractEvent]

Example

>>> all_events = merge_scheduled_and_observed_events(
...     scheduled_events,
...     observed_events
... )

Note

If two events have the same time and type, only the first is kept. This prevents duplicate event processing.

class jactus.contracts.AnnuityContract(*args, **kwargs)[source]

Annuity (ANN) contract.

Amortizing loan with constant total payment amount (principal + interest). The payment amount is automatically calculated using the ACTUS annuity formula to ensure the loan is fully amortized by maturity.

ACTUS Reference:

ACTUS v1.1 Section 7.5 - ANN: Annuity

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize ANN contract.

Parameters:
Raises:
generate_event_schedule()[source]

Generate ANN event schedule per ACTUS specification.

Same as NAM/LAM schedule, with all event types supported.

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize ANN contract state.

Key Feature: If Prnxt is not provided in attributes, calculate it using the ACTUS annuity formula to ensure constant total payments.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get ANN payoff function.

ANN uses the same payoff function as NAM.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

ANN payoff function instance (same as NAM)

Return type:

NAMPayoffFunction

get_state_transition_function(event_type)[source]

Get ANN state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

ANN state transition function instance

Return type:

ANNStateTransitionFunction

class jactus.contracts.CallMoneyContract(*args, **kwargs)[source]

CLM (Call Money) contract implementation.

CLM is an on-demand loan where maturity is determined by observed events rather than fixed at inception. Common for lines of credit and demand loans.

ACTUS Reference:

ACTUS v1.1 Section 7.6 - CLM: Call Money

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CLM contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not CLM

initialize_state()[source]

Initialize CLM contract state.

CLM state is simpler than LAM - no prnxt or ipcb states. When status_date >= IED, the IED event is skipped so state must be initialized from contract attributes directly.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get CLM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

CLM payoff function instance

Return type:

CLMPayoffFunction

get_state_transition_function(event_type)[source]

Get CLM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

CLM state transition function instance

Return type:

CLMStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CLM contract with BDC-aware rate observation.

For SCP/SCF conventions, RR events use the original (unadjusted) schedule date for rate observation, not the BDC-shifted event time.

Parameters:
Return type:

SimulationHistory

generate_event_schedule()[source]

Generate complete event schedule for CLM contract.

CLM schedule is simpler than PAM: - No regular IP events (only at maturity if MD is set) - Optional IPCI events for interest capitalization - Maturity may be dynamic (from observed events)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

class jactus.contracts.CapFloorContract(*args, **kwargs)[source]

Cap-Floor (CAPFL) contract.

An interest rate cap or floor that pays the differential when the floating rate breaches the cap or floor rate.

The contract either: 1. References an underlier via contract_structure (child observer mode) 2. Contains embedded underlier terms for standalone operation

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize base contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes (terms and conditions)

  • risk_factor_observer (RiskFactorObserver) – Observer for accessing market risk factors

  • child_contract_observer (ChildContractObserver | None) – Optional observer for child contracts

  • rngs – Optional Flax RNG state for stochastic contracts

Example

>>> from jactus.observers import ConstantRiskFactorObserver
>>> attrs = ContractAttributes(...)
>>> risk_obs = ConstantRiskFactorObserver(1.0)
>>> contract = MyContract(attrs, risk_obs)
generate_event_schedule()[source]

Generate event schedule for CAPFL contract.

If underlier terms are embedded, generates IP and RR schedules directly from those terms. Otherwise, queries the child observer.

Return type:

EventSchedule

initialize_state()[source]

Initialize CAPFL contract state.

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for a specific event type.

Returns the appropriate payoff function (POF) for calculating the cashflow generated by the given event type.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

PayoffFunction for the event type

Return type:

CapFloorPayoffFunction

Example

>>> def get_payoff_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentPayoff()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionPayoff()
...     else:
...         return ZeroPayoff()

References

ACTUS v1.1 Section 2.7 - Payoff Functions

get_state_transition_function(event_type)[source]

Get state transition function for a specific event type.

Returns the appropriate state transition function (STF) for updating contract state when the given event occurs.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

StateTransitionFunction for the event type

Return type:

CapFloorStateTransitionFunction

Example

>>> def get_state_transition_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentSTF()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionSTF()
...     else:
...         return IdentitySTF()

References

ACTUS v1.1 Section 2.8 - State Transition Functions

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CAPFL contract.

RR events are used internally for rate tracking but filtered from the output since CAPFL only exposes IP events externally.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.CashContract(*args, **kwargs)[source]

Cash (CSH) contract implementation.

Represents a simple cash position with no cashflows. This is the simplest ACTUS contract type, used for cash accounts, settlement amounts, or as initial positions in portfolios.

Characteristics:

  • Only AD (Analysis Date) events

  • Minimal state: notional (nt) and status date (sd)

  • No cashflows (all payoffs = 0.0)

  • No interest accrual or fees

ACTUS Reference:

ACTUS v1.1 Section 7.8 - CSH: Cash

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes including notional, currency, role

risk_factor_observer

Observer for market data (not used in CSH)

child_contract_observer

Observer for child contracts (not used in CSH)

rngs

Random number generators for JAX

Example

>>> from jactus.contracts.csh import CashContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>>
>>> attrs = ContractAttributes(
...     contract_id="CASH-001",
...     contract_type=ContractType.CSH,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     notional_principal=10000.0
... )
>>>
>>> contract = CashContract(
...     attributes=attrs,
...     risk_factor_observer=ConstantRiskFactorObserver(0.0)
... )
>>>
>>> # Simulate the contract
>>> result = contract.simulate()
>>> print(f"Events: {len(result.events)}")  # 1 (just AD)
>>> print(f"Total cashflow: {result.total_cashflow()}")  # 0.0
__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize Cash contract.

Parameters:
Raises:

ValueError – If validation fails

Validations:

  • contract_type must be ContractType.CSH

  • notional_principal (NT) must be defined

  • contract_role (CNTRL) must be defined

  • currency (CUR) must be defined

generate_event_schedule()[source]

Generate CSH event schedule.

CSH contracts only have Analysis Date (AD) events. A single AD event is created at the status date.

Returns:

EventSchedule with one AD event

Return type:

EventSchedule

ACTUS Reference:
CSH Contract Schedule:

  • AD event at status_date

  • No other events

initialize_state()[source]

Initialize CSH contract state.

CSH state is minimal: - nt: Notional with role sign applied: R(CNTRL) × NT - sd: Status date - All other states are zero or null

Returns:

Initial contract state

Return type:

ContractState

ACTUS Reference:
CSH State Initialization:

nt = R(CNTRL) × NT sd = status_date All other states = 0.0 or null

get_payoff_function(event_type)[source]

Get payoff function for CSH events.

Parameters:

event_type (Any) – Type of event (only AD for CSH)

Returns:

CashPayoffFunction instance

Return type:

CashPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CSH events.

Parameters:

event_type (Any) – Type of event (only AD for CSH)

Returns:

CashStateTransitionFunction instance

Return type:

CashStateTransitionFunction

class jactus.contracts.CommodityContract(*args, **kwargs)[source]

Commodity (COM) contract implementation.

Represents a commodity position with price exposure. COM is one of the simplest ACTUS contracts, similar to STK but for physical or financial commodities (gold, oil, wheat, etc.).

ACTUS Reference:

ACTUS v1.1 Section 7.10

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and parameters

risk_factor_observer

Observer for market prices

child_contract_observer

Observer for child contracts (optional)

Example

>>> attrs = ContractAttributes(
...     contract_id="COM-001",
...     contract_type=ContractType.COM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     price_at_purchase_date=7500.0,
...     price_at_termination_date=8200.0,
... )
>>> contract = CommodityContract(attrs, risk_obs)
>>> result = contract.simulate()
__init__(attributes, risk_factor_observer, child_contract_observer=None, rngs=None)[source]

Initialize COM contract.

Parameters:
Raises:

ValueError – If contract_type is not COM or required attributes missing

generate_event_schedule()[source]

Generate COM event schedule.

Generates events for commodity contract: - AD: Analysis dates (if specified) - PRD: Purchase date (if specified) - TD: Termination date (if specified)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

ACTUS Reference:

COM Contract Schedule from Section 7.10

initialize_state()[source]

Initialize COM contract state.

COM has minimal state - only status date and performance.

ACTUS Reference:

COM State Initialization from Section 7.10

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for COM events.

Parameters:

event_type (Any)

Return type:

CommodityPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for COM events.

Parameters:

event_type (Any)

Return type:

CommodityStateTransitionFunction

class jactus.contracts.CreditEnhancementCollateralContract(*args, **kwargs)[source]

Credit Enhancement Collateral (CEC) contract.

A collateral contract that tracks collateral value from covering contracts against exposure from covered contracts. When credit events occur or at maturity, compares collateral vs required amount and settles appropriately.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for covered/covering contracts (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CEC contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for CEC contract.

The schedule includes: 1. Analysis dates (AD) if specified 2. Credit event detection (XD) if covered contract defaults 3. Settlement event (STD) after credit event or at maturity 4. Maturity event (MD) if no credit event occurs

Returns:

EventSchedule with collateral events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State includes collateral value compared to required amount.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for CEC contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CECPayoffFunction instance

Return type:

CECPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CEC contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CECStateTransitionFunction instance

Return type:

CECStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CEC contract with comprehensive event generation.

Generates XD, STD, and MD events based on covered/covering contract states and credit events observed through the child observer.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.CreditEnhancementGuaranteeContract(*args, **kwargs)[source]

Credit Enhancement Guarantee (CEG) contract.

A guarantee contract that pays out when covered contracts experience credit events. The payout covers a specified percentage of the covered amount, calculated based on the coverage extent mode (notional only, notional plus interest, or market value).

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for covered contract data (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CEG contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for CEG contract.

The schedule includes: 1. Fee payment events (FP) if fees are charged 2. Credit event detection (XD) if covered contract defaults 3. Settlement event (STD) after credit event 4. Maturity event (MD) if no credit event occurs

Returns:

EventSchedule with guarantee events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State includes coverage amount calculated from covered contracts.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for CEG contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CEGPayoffFunction instance

Return type:

CEGPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CEG contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CEGStateTransitionFunction instance

Return type:

CEGStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CEG contract with comprehensive event generation.

Generates PRD, FP, XD, STD, and MD events based on covered contract states and credit events observed through the child observer.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.ExoticLinearAmortizerContract(*args, **kwargs)[source]

LAX (Exotic Linear Amortizer) contract implementation.

LAX is the most complex amortizing contract, supporting flexible array schedules for principal redemption, interest payments, and rate resets.

ACTUS Reference:

ACTUS v1.1 Section 7.3 - LAX: Exotic Linear Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize LAX contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not LAX

initialize_state()[source]

Initialize LAX contract state.

Initializes all state variables including Prnxt and Ipcb (same as LAM).

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get LAX payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

LAX payoff function instance

Return type:

LAXPayoffFunction

get_state_transition_function(event_type)[source]

Get LAX state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

LAX state transition function instance

Return type:

LAXStateTransitionFunction

generate_event_schedule()[source]

Generate complete event schedule for LAX contract.

LAX uses array schedules to generate PR, PI, PRF, IP, RR, and RRF events.

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate LAX contract with array-aware prnxt injection.

Before each PR/PI event, updates state.prnxt from the array schedule so the correct principal amount is used without explicit PRF events.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.FutureContract(*args, **kwargs)[source]

Future Contract (FUTUR) implementation.

Represents a futures contract with linear payoff based on spot vs futures price.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

future_price

Agreed futures price

Type:

PFUT

contract_structure

Underlier reference

Type:

CTST

notional_principal

Number of units (e.g., contracts)

Type:

NT

maturity_date

Futures expiration date

Type:

MD

Key Differences from Options:

  • Linear payoff (not max-based)

  • No premium payment

  • Settlement amount can be negative

  • Always settles at maturity (no exercise decision)

Example

>>> # S&P 500 futures contract
>>> attrs = ContractAttributes(
...     contract_type=ContractType.FUTUR,
...     future_price=4500.0,
...     contract_structure="SPX",
...     notional_principal=1.0,
...     maturity_date=ActusDateTime(2024, 12, 31, 0, 0, 0),
...     ...
... )
>>> contract = FutureContract(attrs, rf_obs)
>>> events = contract.generate_event_schedule()
__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize FUTUR contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate event schedule for FUTUR contract.

Events: - AD (optional): Analysis dates - PRD (optional): Purchase date (zero payment) - TD (optional): Termination date - MD: Maturity date (settlement amount calculated) - XD: Exercise date at maturity - STD: Settlement date (receive settlement amount) - Pre-exercised (exercise_date + exercise_amount set): STD only

Returns:

Event schedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial contract state with xa set to exercise_amount if pre-exercised

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for FUTUR contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

FuturePayoffFunction instance

Return type:

FuturePayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for FUTUR contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

FutureStateTransitionFunction instance

Return type:

FutureStateTransitionFunction

class jactus.contracts.FXOutrightContract(*args, **kwargs)[source]

Foreign Exchange Outright (FXOUT) contract.

Represents an FX forward or spot contract with settlement in two currencies.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT: Foreign Exchange Outright

Key Attributes:

currency (CUR): First currency (e.g., “EUR”) currency_2 (CUR2): Second currency (e.g., “USD”) notional_principal (NT): Amount in first currency notional_principal_2 (NT2): Amount in second currency delivery_settlement (DS): ‘D’ (delivery/net) or ‘S’ (dual/gross) maturity_date (MD) or settlement_date (STD): Settlement date

Settlement Modes:
Delivery (DS=’D’):

  • Single STD event with net settlement

  • Payoff = NT - (FX_rate × NT2)

  • Settled in first currency

Dual (DS=’S’):

  • Two STD events: STD(1) and STD(2)

  • STD(1): Receive NT in first currency

  • STD(2): Pay NT2 in second currency

  • Full principal exchange

State Variables:

md: Maturity/settlement date prf: Contract performance sd: Status date

Example

EUR/USD forward contract: - Buy 100,000 EUR - Sell 110,000 USD - Forward rate = 1.10 (agreed) - At maturity, observe market rate - If market rate = 1.12, profit = 100,000 - (100,000 × 110,000/112,000) ≈ 1,786 EUR

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize FXOUT contract.

Parameters:
Raises:

ValueError – If contract type is not FXOUT or required attributes missing

generate_event_schedule()[source]

Generate event schedule for FXOUT contract.

Events depend on delivery/settlement mode and settlement period: - DS=’S’ with settlement period → single STD at maturity + SP (net settlement) - Otherwise → MD events at maturity (gross exchange, two currency legs) - Early termination (TD < maturity) → suppress MD/STD events

Returns:

EventSchedule with contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT State Variables Initialization

State Variables:

tmd: MD if STD = ∅, else STD prf: PRF (contract performance) sd: Status date nt, ipnr, ipac, feac, nsc, isc: Zero/default values (not used by FXOUT)

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for FXOUT contract.

Parameters:

event_type (Any) – Event type (not used for FXOUT, same function for all events)

Returns:

FXOutrightPayoffFunction instance

Return type:

BasePayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for FXOUT contract.

Parameters:

event_type (Any) – Event type (not used for FXOUT, same function for all events)

Returns:

FXOutrightStateTransitionFunction instance

Return type:

BaseStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate FXOUT contract with dual-currency MD and net STD handling.

Overrides base simulate() to compute: - Per-leg payoffs for gross settlement (MD events) - Net settlement payoff for cash settlement (STD events)

Parameters:
Return type:

SimulationHistory

class jactus.contracts.GenericSwapContract(*args, **kwargs)[source]

Generic Swap (SWAPS) contract.

A swap with two explicit child contract legs. Supports any contract types for the legs (PAM, LAM, ANN, etc.) and provides flexible event merging and state aggregation.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for child contract data (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize SWAPS contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for SWAPS contract.

The schedule is created by: 1. Querying events from both child legs 2. Merging congruent events if DS=’D’ (net settlement) 3. Keeping all events separate if DS=’S’ (gross settlement)

Returns:

EventSchedule with merged or separate leg events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State is aggregated from both child leg states.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for SWAPS contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

GenericSwapPayoffFunction instance

Return type:

GenericSwapPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for SWAPS contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

GenericSwapStateTransitionFunction instance

Return type:

GenericSwapStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate SWAPS contract by passing through child contract events.

For SWAPS, event payoffs and states come directly from child contract simulations. The schedule events already contain pre-computed data, so we pass them through instead of recalculating via POF/STF.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.LinearAmortizerContract(*args, **kwargs)[source]

Linear Amortizer (LAM) contract.

Amortizing loan with fixed principal redemption amounts. Principal is repaid in regular installments (PR events), with interest calculated on an Interest Calculation Base (IPCB) that can track current notional, stay fixed, or update periodically.

ACTUS Reference:

ACTUS v1.1 Section 7.2 - LAM: Linear Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize LAM contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:
generate_event_schedule()[source]

Generate LAM event schedule per ACTUS specification.

Schedule formula for each event type:

IED: Single event at initial_exchange_date (if IED >= SD) PR: S(PRANX, PRCL, MD) - principal redemption schedule IP: S(IPANX, IPCL, MD) - interest payment schedule IPCI: S(IPANX, IPCL, IPCED) - interest capitalization until end date IPCB: S(IPCBANX, IPCBCL, MD) - if IPCB=’NTL’ RR: S(RRANX, RRCL, MD) - rate reset schedule FP: S(FEANX, FECL, MD) - fee payment schedule PRD: Single event at purchase_date TD: Single event at termination_date (truncates schedule) MD: Single event at maturity_date

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize LAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran: Nt, Ipnr are set from attributes, and interest is accrued from IED (or IPANX) to SD.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get LAM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

LAM payoff function instance

Return type:

LAMPayoffFunction

get_state_transition_function(event_type)[source]

Get LAM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

LAM state transition function instance

Return type:

LAMStateTransitionFunction

class jactus.contracts.NegativeAmortizerContract(*args, **kwargs)[source]

Negative Amortizer (NAM) contract.

Amortizing loan where principal can increase when payments are less than accrued interest (negative amortization). Extends LAM pattern with modified principal redemption handling.

ACTUS Reference:

ACTUS v1.1 Section 7.4 - NAM: Negative Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize NAM contract.

Parameters:
Raises:
generate_event_schedule()[source]

Generate NAM event schedule per ACTUS specification.

Same as LAM schedule, with all event types supported.

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize NAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get NAM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

NAM payoff function instance

Return type:

NAMPayoffFunction

get_state_transition_function(event_type)[source]

Get NAM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

NAM state transition function instance

Return type:

NAMStateTransitionFunction

class jactus.contracts.OptionContract(*args, **kwargs)[source]

Option Contract (OPTNS) implementation.

Represents a vanilla option contract with call, put, or collar payoffs. Supports European, American, and Bermudan exercise styles.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

option_type

‘C’ (call), ‘P’ (put), ‘CP’ (collar)

Type:

OPTP

option_strike_1

Primary strike price

Type:

OPS1

option_strike_2

Secondary strike (collar only)

Type:

OPS2

option_exercise_type

‘E’ (European), ‘A’ (American), ‘B’ (Bermudan)

Type:

OPXT

contract_structure

Underlier reference

Type:

CTST

notional_principal

Number of units (e.g., shares)

Type:

NT

price_at_purchase_date

Premium per unit

Type:

PPRD

maturity_date

Option expiration date

Type:

MD

settlement_period

Period from exercise to settlement

Type:

STPD

Example

>>> # European call option
>>> attrs = ContractAttributes(
...     contract_type=ContractType.OPTNS,
...     option_type="C",
...     option_strike_1=100.0,
...     option_exercise_type="E",
...     contract_structure="AAPL",
...     ...
... )
>>> contract = OptionContract(attrs, rf_obs)
>>> events = contract.generate_event_schedule()
__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize OPTNS contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate event schedule for OPTNS contract.

Events depend on exercise type: - European: AD (optional), PRD, MD, XD, STD - American: AD (optional), PRD, XD (multiple), MD, STD - Bermudan: AD (optional), PRD, XD (specific dates), MD, STD - Pre-exercised (exercise_date + exercise_amount set): STD only

Returns:

Event schedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial contract state with xa set to exercise_amount if pre-exercised

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for OPTNS contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

OptionPayoffFunction instance

Return type:

OptionPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for OPTNS contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

OptionStateTransitionFunction instance

Return type:

OptionStateTransitionFunction

class jactus.contracts.PlainVanillaSwapContract(*args, **kwargs)[source]

Plain Vanilla Interest Rate Swap (SWPPV) contract.

Swaps fixed and floating interest rate payments on a notional amount. No exchange of principal occurs.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize SWPPV contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for SWPPV contract.

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for SWPPV contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

PlainVanillaSwapPayoffFunction instance

Return type:

PlainVanillaSwapPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for SWPPV contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

PlainVanillaSwapStateTransitionFunction instance

Return type:

PlainVanillaSwapStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate SWPPV contract.

Overrides base to filter out events before purchaseDate and after terminationDate. The full event schedule is processed for state computation, but only visible events are returned.

Parameters:
Return type:

SimulationHistory

class jactus.contracts.PrincipalAtMaturityContract(*args, **kwargs)[source]

Principal At Maturity (PAM) contract implementation.

Represents a bullet loan where principal is repaid in full at maturity with regular interest payments.

ACTUS Reference:

ACTUS v1.1 Section 7.1 - PAM: Principal At Maturity

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Observer for market data

child_contract_observer

Observer for child contracts

rngs

Random number generators for JAX

__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize PAM contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate PAM event schedule per ACTUS specification.

Schedule formula for each event type:

IED: Single event at initial_exchange_date (if IED >= SD) IP: S(IPANX, IPCL, MD) - from interest payment anchor IPCI: S(IPANX, IPCL, IPCED) - interest capitalization until end date RR: S(RRANX, RRCL, MD) - rate reset schedule RRF: S(RRANX, RRCL, MD) - if rateResetFixing defined FP: S(FEANX, FECL, MD) - fee payment schedule PRD: Single event at purchase_date TD: Single event at termination_date (truncates schedule) MD: Single event at maturity_date

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize PAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran: Nt, Ipnr are set from attributes, and interest is accrued from IED (or IPANX) to SD.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for PAM events.

Parameters:

event_type (Any)

Return type:

PAMPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for PAM events.

Parameters:

event_type (Any)

Return type:

PAMStateTransitionFunction

class jactus.contracts.StockContract(*args, **kwargs)[source]

Stock (STK) contract implementation.

Represents an equity position with potential dividend payments. STK is one of the simplest ACTUS contracts, with minimal state and straightforward cashflow logic.

ACTUS Reference:

ACTUS v1.1 Section 7.9

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and parameters

risk_factor_observer

Observer for market prices and dividends

child_contract_observer

Observer for child contracts (optional)

Example

>>> attrs = ContractAttributes(
...     contract_id="STK-001",
...     contract_type=ContractType.STK,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     market_object_code="AAPL",
... )
>>> contract = StockContract(attrs, risk_obs)
>>> result = contract.simulate()
__init__(attributes, risk_factor_observer, child_contract_observer=None, rngs=None)[source]

Initialize STK contract.

Parameters:
Raises:

ValueError – If contract_type is not STK or required attributes missing

generate_event_schedule()[source]

Generate STK event schedule.

Generates events for stock contract: - AD: Analysis dates (if specified) - PRD: Purchase date (if specified) - TD: Termination date (if specified) - DV: Dividend events (if dividend schedule specified)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

ACTUS Reference:

STK Contract Schedule from Section 7.9

initialize_state()[source]

Initialize STK contract state.

STK has minimal state - only status date and performance.

ACTUS Reference:

STK State Initialization from Section 7.9

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for STK events.

Parameters:

event_type (Any)

Return type:

StockPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for STK events.

Parameters:

event_type (Any)

Return type:

StockStateTransitionFunction

class jactus.contracts.UndefinedMaturityProfileContract(*args, **kwargs)[source]

Undefined Maturity Profile (UMP) contract.

UMP represents a loan with uncertain principal repayment schedule. All principal changes (PR/PI events) come from observed events rather than a fixed schedule. This is useful for modeling: - Lines of credit with uncertain drawdowns/repayments - Loans with irregular principal payments - Cashflow profiles determined by external events

Key Characteristics:

  • All PR events from observer (no fixed schedule)

  • No IP schedule (only IPCI for capitalization)

  • Uncertain cashflows (all principal from observations)

  • Maturity from last observed event

  • Simpler state than LAM (no prnxt or ipcb)

Event Types:

  • AD: Analysis Date

  • IED: Initial Exchange

  • PR: Principal Repayment (from observer)

  • PI: Principal Increase (from observer)

  • FP: Fee Payment

  • IPCI: Interest Capitalization

  • RR: Rate Reset

  • RRF: Rate Reset Fixing

  • CE: Credit Event

  • MD: Maturity

State Variables:

  • sd: Status Date

  • tmd: Terminal Maturity Date (from observed events)

  • nt: Notional

  • ipnr: Interest Rate

  • ipac: Accrued Interest

  • feac: Accrued Fees

  • nsc: Notional Scaling

  • isc: Interest Scaling

Example

>>> attrs = ContractAttributes(
...     contract_id="UMP-001",
...     contract_type=ContractType.UMP,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=100000.0,
...     nominal_interest_rate=0.06,
...     ipci_cycle="1Q",  # Quarterly capitalization
... )
>>> contract = UndefinedMaturityProfileContract(attrs)

References

ACTUS v1.1 Section 7.7 - Undefined Maturity Profile

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize UMP contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not UMP

initialize_state()[source]

Initialize UMP state.

UMP state is simpler than LAM - no prnxt or ipcb states. Maturity is determined from observed events.

Return type:

ContractState

get_payoff_function(event_type)[source]

Get UMP payoff function.

Parameters:

event_type (Any) – Event type (unused - same POF for all)

Returns:

UMP payoff function instance

Return type:

UMPPayoffFunction

get_state_transition_function(event_type)[source]

Get UMP state transition function.

Parameters:

event_type (Any) – Event type (unused - same STF for all)

Returns:

UMP state transition function instance

Return type:

UMPStateTransitionFunction

generate_event_schedule()[source]

Generate UMP event schedule.

UMP schedule includes: - AD: Analysis date (if provided) - IED: Initial exchange - IPCI: Interest capitalization (from interest_payment_cycle) - RR: Rate reset (if RR cycle provided) - RRF: Rate reset fixing (if RRF cycle provided) - FP: Fee payment (if FP cycle provided) - TD: Termination (if termination_date is set) - MD: Maturity (from observed events or attributes)

Note: PR/PI events come from observer, not from schedule.

Returns:

Event schedule with all scheduled events

Return type:

EventSchedule

jactus.contracts.create_contract(attributes, risk_factor_observer, child_contract_observer=None)[source]

Create a contract instance using the factory pattern.

Dynamically instantiates the correct contract class based on the contract_type attribute. This is the recommended way to create contracts when the type is determined at runtime.

Parameters:
Returns:

Instance of the appropriate contract class

Raises:
Return type:

BaseContract

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="PAM-001",
...     contract_type=ContractType.PAM,
...     notional_principal=100000.0,
...     currency="USD",
... )
>>> rf_obs = ConstantRiskFactorObserver(0.05)
>>> contract = create_contract(attrs, rf_obs)
>>> isinstance(contract, PrincipalAtMaturityContract)
True
jactus.contracts.register_contract_type(contract_type, contract_class)[source]

Register a new contract type in the factory registry.

This allows for dynamic extensibility - users can register custom contract implementations without modifying the core library.

Parameters:

  • contract_type (ContractType) – The ContractType enum value

  • contract_class (type[BaseContract]) – The contract implementation class (must extend BaseContract)

Raises:

  • TypeError – If contract_class doesn’t extend BaseContract

  • ValueError – If contract_type is already registered

Return type:

None

Example

>>> class MyCustomContract(BaseContract):
...     # Custom implementation
...     pass
>>>
>>> register_contract_type(ContractType.CUSTOM, MyCustomContract)
jactus.contracts.get_available_contract_types()[source]

Get list of all registered contract types.

Returns:

List of ContractType enum values that are currently registered

Return type:

list[ContractType]

Example

>>> types = get_available_contract_types()
>>> ContractType.CSH in types
True
>>> ContractType.PAM in types
True
class jactus.contracts.PAMArrayState(nt, ipnr, ipac, feac, nsc, isc)[source]

Minimal scan-loop state for PAM simulation.

All fields are scalar jnp.ndarray (float32). sd (status date) is omitted because year fractions are pre-computed before the JIT boundary.

Parameters:
nt: Array

Alias for field number 0

ipnr: Array

Alias for field number 1

ipac: Array

Alias for field number 2

feac: Array

Alias for field number 3

nsc: Array

Alias for field number 4

isc: Array

Alias for field number 5

class jactus.contracts.PAMArrayParams(role_sign, notional_principal, nominal_interest_rate, premium_discount_at_ied, accrued_interest, fee_rate, fee_basis, penalty_rate, penalty_type, price_at_purchase_date, price_at_termination_date, rate_reset_spread, rate_reset_multiplier, rate_reset_floor, rate_reset_cap, rate_reset_next, has_rate_floor, has_rate_cap, ied_ipac)[source]

Static contract parameters extracted from ContractAttributes.

These do not change during the scan loop. Enum-based branches (penalty type, fee basis) are encoded as integers for jnp.where.

Parameters:
role_sign: Array

Alias for field number 0

notional_principal: Array

Alias for field number 1

nominal_interest_rate: Array

Alias for field number 2

premium_discount_at_ied: Array

Alias for field number 3

accrued_interest: Array

Alias for field number 4

fee_rate: Array

Alias for field number 5

fee_basis: Array

Alias for field number 6

penalty_rate: Array

Alias for field number 7

penalty_type: Array

Alias for field number 8

price_at_purchase_date: Array

Alias for field number 9

price_at_termination_date: Array

Alias for field number 10

rate_reset_spread: Array

Alias for field number 11

rate_reset_multiplier: Array

Alias for field number 12

rate_reset_floor: Array

Alias for field number 13

rate_reset_cap: Array

Alias for field number 14

rate_reset_next: Array

Alias for field number 15

has_rate_floor: Array

Alias for field number 16

has_rate_cap: Array

Alias for field number 17

ied_ipac: Array

Alias for field number 18

jactus.contracts.simulate_pam_array(initial_state, event_types, year_fractions, rf_values, params)[source]

Run a PAM simulation as a pure JAX function.

This function is JIT-compilable and vmap-able.

Parameters:

  • initial_state (PAMArrayState) – Starting state (6 scalar fields).

  • event_types (Array) – (num_events,) int32 — EventType.index values.

  • year_fractions (Array) – (num_events,) float32 — pre-computed YF per event.

  • rf_values (Array) – (num_events,) float32 — pre-computed risk factor per event.

  • params (PAMArrayParams) – Static contract parameters.

Returns:

(final_state, payoffs) where payoffs is (num_events,) float32.

Return type:

tuple[PAMArrayState, Array]

jactus.contracts.simulate_pam_array_jit(initial_state, event_types, year_fractions, rf_values, params)

Run a PAM simulation as a pure JAX function.

This function is JIT-compilable and vmap-able.

Parameters:

  • initial_state (PAMArrayState) – Starting state (6 scalar fields).

  • event_types (Array) – (num_events,) int32 — EventType.index values.

  • year_fractions (Array) – (num_events,) float32 — pre-computed YF per event.

  • rf_values (Array) – (num_events,) float32 — pre-computed risk factor per event.

  • params (PAMArrayParams) – Static contract parameters.

Returns:

(final_state, payoffs) where payoffs is (num_events,) float32.

Return type:

tuple[PAMArrayState, Array]

jactus.contracts.batch_precompute_pam(params, max_ip)[source]

JAX-native batch schedule generation + year fractions.

Wraps a JIT-compiled kernel. max_ip is a compile-time constant (recompiles only when max_ip changes).

Parameters:

  • params (BatchContractParams) – Batch contract parameters (shape (N,) per field).

  • max_ip (int) – Maximum IP events (static, determines array shapes).

Returns:

(event_types, year_fractions, rf_values, masks) — all shape (N, max_events) where max_events = max_ip + 3.

Return type:

tuple[Array, Array, Array, Array]

jactus.contracts.batch_simulate_pam(initial_states, event_types, year_fractions, rf_values, params)[source]

Batched PAM simulation without vmap — single scan over [B] arrays.

Eliminates JAX vmap CPU dispatch overhead by operating directly on batch-dimensioned arrays. Each scan step computes all event-type outcomes for all contracts simultaneously using branchless jnp.where dispatch.

Parameters:

  • initial_states (PAMArrayState) – PAMArrayState with each field shape [B].

  • event_types (Array) – [B, T] int32 — event type indices per contract.

  • year_fractions (Array) – [B, T] float32.

  • rf_values (Array) – [B, T] float32.

  • params (PAMArrayParams) – PAMArrayParams with each field shape [B].

Returns:

(final_states, payoffs) where payoffs is [B, T].

Return type:

tuple[PAMArrayState, Array]

jactus.contracts.batch_simulate_pam_auto(initial_states, event_types, year_fractions, rf_values, params)[source]

Batched simulation using the optimal strategy for all backends.

Uses the single-scan batch approach (batch_simulate_pam) which processes all contracts in shaped [B, T] arrays via a single lax.scan. This is faster than vmap on CPU, GPU, and TPU because it avoids per-contract dispatch overhead.

The vmap variant (batch_simulate_pam_vmap) remains available for explicit use but is not selected automatically.

Parameters:
Return type:

tuple[PAMArrayState, Array]

jactus.contracts.batch_simulate_pam_vmap(initial_state, event_types, year_fractions, rf_values, params)

Vectorized version of simulate_pam_array. Takes similar arguments as simulate_pam_array but with additional array axes over which simulate_pam_array is mapped.

Original documentation:

Run a PAM simulation as a pure JAX function.

This function is JIT-compilable and vmap-able.

Args:

initial_state: Starting state (6 scalar fields). event_types: (num_events,) int32 — EventType.index values. year_fractions: (num_events,) float32 — pre-computed YF per event. rf_values: (num_events,) float32 — pre-computed risk factor per event. params: Static contract parameters.

Returns:

(final_state, payoffs) where payoffs is (num_events,) float32.

Parameters:
Return type:

tuple[PAMArrayState, Array]

jactus.contracts.precompute_pam_arrays(attrs, rf_observer)[source]

Pre-compute JAX arrays for array-mode PAM simulation.

Generates the event schedule and initial state directly from attributes (bypassing PrincipalAtMaturityContract), then converts to JAX arrays suitable for simulate_pam_array.

Parameters:
Returns:

(initial_state, event_types, year_fractions, rf_values, params)

Return type:

tuple[PAMArrayState, Array, Array, Array, PAMArrayParams]

jactus.contracts.prepare_pam_batch(contracts)[source]

Pre-compute and pad arrays for a batch of PAM contracts.

When _USE_BATCH_SCHEDULE is enabled, eligible contracts have their schedules and year fractions generated via a JAX-native batch path (GPU/TPU-ready). Ineligible contracts fall back to per-contract Python pre-computation.

When all contracts are batch-eligible, a fast path avoids the JAX→NumPy→JAX round-trip, keeping schedule arrays on-device.

Parameters:

contracts (list[tuple[ContractAttributes, RiskFactorObserver]]) – List of (attributes, rf_observer) pairs.

Returns:

(initial_states, event_types, year_fractions, rf_values, params, masks) where each array has a leading batch dimension.

Return type:

tuple[PAMArrayState, Array, Array, Array, PAMArrayParams, Array]

jactus.contracts.simulate_pam_portfolio(contracts, discount_rate=None, year_fractions_from_valuation=None)[source]

End-to-end portfolio simulation with optional PV.

Parameters:

  • contracts (list[tuple[ContractAttributes, RiskFactorObserver]]) – List of (attributes, rf_observer) pairs.

  • discount_rate (float | None) – If provided, compute present values.

  • year_fractions_from_valuation (Array | None) – (batch, max_events) year fractions from valuation date for PV discounting. If None and discount_rate is set, year fractions are computed from each contract’s status_date.

Returns:

Dict with payoffs, masks, final_states, and optionally present_values and total_pv.

Return type:

dict[str, Any]

jactus.contracts.simulate_portfolio(contracts, discount_rate=None)[source]

Simulate a mixed-type portfolio using optimal batch strategies.

Groups contracts by type, dispatches to each type’s batch kernel, and falls back to the scalar Python path for unsupported types.

Parameters:

  • contracts (list[tuple[ContractAttributes, RiskFactorObserver]]) – List of (attributes, rf_observer) pairs. Contract types may be mixed.

  • discount_rate (float | None) – If provided, compute present values (passed to each type’s portfolio function where supported).

Returns:

  • total_cashflows: jnp.ndarray of shape (N,) — total cashflow for each contract in input order.

  • num_contracts: Total number of contracts.

  • batch_contracts: Number of contracts simulated via batch kernels.

  • fallback_contracts: Number of contracts simulated via the scalar Python path.

  • types_used: Set of ContractType values present.

  • per_type_results: Dict mapping ContractType to the raw result dict from each type’s portfolio function (only for batch-simulated types).

Return type:

Dict with

Principal Contracts

PAM - Principal at Maturity

Principal At Maturity (PAM) contract implementation.

This module implements the PAM contract type - a bullet loan with interest payments where the principal is repaid at maturity. PAM is the foundational loan contract and serves as a template for amortizing contracts (LAM, NAM, ANN).

ACTUS Reference:

ACTUS v1.1 Section 7.1 - PAM: Principal At Maturity

Key Features:

  • Principal repaid in full at maturity

  • Regular interest payments (IP events)

  • Optional interest capitalization (IPCI)

  • Variable interest rates with rate resets (RR, RRF)

  • Fees (FP events)

  • Prepayments (PP events)

  • Scaling (SC events)

  • 14 event types total

Example

>>> from jactus.contracts.pam import PrincipalAtMaturityContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="LOAN-001",
...     contract_type=ContractType.PAM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=100000.0,
...     nominal_interest_rate=0.05,
...     day_count_convention=DayCountConvention.A360,
...     interest_payment_cycle="1Y"
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.05)
>>> contract = PrincipalAtMaturityContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.pam.PAMPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for PAM contracts.

Implements all 14 PAM payoff functions according to ACTUS specification. Uses dictionary-based dispatch for O(1) lookup and future jax.lax.switch compatibility (requires EventType.index integer mapping).

ACTUS Reference:

ACTUS v1.1 Section 7.1 - PAM Payoff Functions

Events:

AD: Analysis Date (0.0) IED: Initial Exchange Date (disburse principal) MD: Maturity Date (return principal + accrued) PP: Principal Prepayment PY: Penalty Payment FP: Fee Payment PRD: Purchase Date TD: Termination Date IP: Interest Payment IPCI: Interest Capitalization RR: Rate Reset RRF: Rate Reset Fixing SC: Scaling CE: Credit Event

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for PAM events.

Dispatches to specific payoff function via dict lookup (O(1)).

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

ACTUS Reference:

POF_[event]_PAM functions from Section 7.1

class jactus.contracts.pam.PAMStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for PAM contracts.

Implements all 14 PAM state transition functions according to ACTUS specification. Uses dictionary-based dispatch for O(1) lookup.

ACTUS Reference:

ACTUS v1.1 Section 7.1 - PAM State Transition Functions

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Transition PAM contract state.

Dispatches to specific state transition function via dict lookup (O(1)).

Parameters:
Returns:

Updated contract state

Return type:

ContractState

ACTUS Reference:

STF_[event]_PAM functions from Section 7.1

class jactus.contracts.pam.PrincipalAtMaturityContract(*args, **kwargs)[source]

Bases: BaseContract

Principal At Maturity (PAM) contract implementation.

Represents a bullet loan where principal is repaid in full at maturity with regular interest payments.

ACTUS Reference:

ACTUS v1.1 Section 7.1 - PAM: Principal At Maturity

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Observer for market data

child_contract_observer

Observer for child contracts

rngs

Random number generators for JAX

__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize PAM contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate PAM event schedule per ACTUS specification.

Schedule formula for each event type:

IED: Single event at initial_exchange_date (if IED >= SD) IP: S(IPANX, IPCL, MD) - from interest payment anchor IPCI: S(IPANX, IPCL, IPCED) - interest capitalization until end date RR: S(RRANX, RRCL, MD) - rate reset schedule RRF: S(RRANX, RRCL, MD) - if rateResetFixing defined FP: S(FEANX, FECL, MD) - fee payment schedule PRD: Single event at purchase_date TD: Single event at termination_date (truncates schedule) MD: Single event at maturity_date

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize PAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran: Nt, Ipnr are set from attributes, and interest is accrued from IED (or IPANX) to SD.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for PAM events.

Parameters:

event_type (Any)

Return type:

PAMPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for PAM events.

Parameters:

event_type (Any)

Return type:

PAMStateTransitionFunction

LAM - Linear Amortizer

Linear Amortizer (LAM) contract implementation.

This module implements the LAM contract type - an amortizing loan with fixed principal redemption amounts where principal is repaid in regular installments. LAM is the foundation for other amortizing contracts (NAM, ANN, LAX).

ACTUS Reference:

ACTUS v1.1 Section 7.2 - LAM: Linear Amortizer

Key Features:

  • Fixed principal redemption amounts (Prnxt)

  • Regular principal reduction (PR events)

  • Interest calculated on IPCB (Interest Calculation Base)

  • Three IPCB modes: NT (notional tracking), NTIED (fixed at IED), NTL (lagged)

  • Optional IPCB events for base fixing

  • Maturity can be calculated if not provided

  • 16 event types total

IPCB Modes:

  • NT: Interest calculated on current notional (lagging one period)

  • NTIED: Interest calculated on initial notional at IED (fixed)

  • NTL: Interest calculated on notional at last IPCB event (lagged with updates)

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.core import ActusDateTime, DayCountConvention
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="MORTGAGE-001",
...     contract_type=ContractType.LAM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2054, 1, 15, 0, 0, 0),  # 30 years
...     currency="USD",
...     notional_principal=300000.0,
...     nominal_interest_rate=0.065,
...     day_count_convention=DayCountConvention.A360,
...     principal_redemption_cycle="1M",  # Monthly payments
...     next_principal_redemption_amount=1000.0,  # $1000/month principal
...     interest_calculation_base="NT"  # Interest on current notional
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.065)
>>> contract = create_contract(attrs, rf_obs)
>>> result = contract.simulate()
class jactus.contracts.lam.LAMPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for LAM contracts.

Implements all LAM payoff functions according to ACTUS specification. The key difference from PAM is the PR (Principal Redemption) event and the use of IPCB (Interest Calculation Base) for interest calculations.

ACTUS Reference:

ACTUS v1.1 Section 7.2 - LAM Payoff Functions

Events:

AD: Analysis Date (0.0) IED: Initial Exchange Date (disburse principal) PR: Principal Redemption (fixed principal payment) MD: Maturity Date (final principal + interest) PP: Principal Prepayment PY: Penalty Payment FP: Fee Payment PRD: Purchase Date TD: Termination Date IP: Interest Payment (on IPCB, not current notional) IPCI: Interest Capitalization IPCB: Interest Calculation Base fixing RR: Rate Reset RRF: Rate Reset Fixing SC: Scaling CE: Credit Event

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for LAM event via dict dispatch.

Parameters:
Returns:

JAX array containing the payoff amount

Return type:

Array

class jactus.contracts.lam.LAMStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for LAM contracts.

Implements all LAM state transitions according to ACTUS specification. The key differences from PAM are PR event handling and IPCB tracking.

ACTUS Reference:

ACTUS v1.1 Section 7.2 - LAM State Transition Functions

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer)[source]

Apply state transition for LAM event via dict dispatch.

Parameters:
Returns:

New contract state after event

Return type:

ContractState

class jactus.contracts.lam.LinearAmortizerContract(*args, **kwargs)[source]

Bases: BaseContract

Linear Amortizer (LAM) contract.

Amortizing loan with fixed principal redemption amounts. Principal is repaid in regular installments (PR events), with interest calculated on an Interest Calculation Base (IPCB) that can track current notional, stay fixed, or update periodically.

ACTUS Reference:

ACTUS v1.1 Section 7.2 - LAM: Linear Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize LAM contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:
generate_event_schedule()[source]

Generate LAM event schedule per ACTUS specification.

Schedule formula for each event type:

IED: Single event at initial_exchange_date (if IED >= SD) PR: S(PRANX, PRCL, MD) - principal redemption schedule IP: S(IPANX, IPCL, MD) - interest payment schedule IPCI: S(IPANX, IPCL, IPCED) - interest capitalization until end date IPCB: S(IPCBANX, IPCBCL, MD) - if IPCB=’NTL’ RR: S(RRANX, RRCL, MD) - rate reset schedule FP: S(FEANX, FECL, MD) - fee payment schedule PRD: Single event at purchase_date TD: Single event at termination_date (truncates schedule) MD: Single event at maturity_date

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize LAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran: Nt, Ipnr are set from attributes, and interest is accrued from IED (or IPANX) to SD.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get LAM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

LAM payoff function instance

Return type:

LAMPayoffFunction

get_state_transition_function(event_type)[source]

Get LAM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

LAM state transition function instance

Return type:

LAMStateTransitionFunction

LAX - Exotic Linear Amortizer

Exotic Linear Amortizer (LAX) contract implementation.

This module implements the LAX contract type - the most complex amortizing contract with flexible array schedules that allow varying principal redemption amounts, rates, and cycles over the life of the contract.

ACTUS Reference:

ACTUS v1.1 Section 7.3 - LAX: Exotic Linear Amortizer

Key Features:

  • Array schedules for principal redemption (ARPRANX, ARPRCL, ARPRNXT)

  • Array schedules for interest payments (ARIPANX, ARIPCL)

  • Array schedules for rate resets (ARRRANX, ARRRCL, ARRATE)

  • Increase/decrease indicators (ARINCDEC) for principal changes

  • Fixed/variable rate indicators (ARFIXVAR)

  • PI (Principal Increase) and PR (Principal Redemption) events

  • PRF (Principal Redemption Amount Fixing) events

  • All IPCB modes from LAM

Array Schedule Concept:

Instead of a single cycle and anchor, LAX uses arrays to define multiple sub-schedules with different parameters. For example: - ARPRANX = [2024-01-15, 2025-01-15, 2026-01-15] - ARPRCL = [“1M”, “1M”, “1M”] - ARPRNXT = [1000, 2000, 3000] - ARINCDEC = [“INC”, “INC”, “DEC”] This creates increasing principal for 2 years, then decreasing.

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.core import ActusDateTime, DayCountConvention
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="STEP-UP-LOAN-001",
...     contract_type=ContractType.LAX,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2027, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=100000.0,
...     nominal_interest_rate=0.05,
...     day_count_convention=DayCountConvention.A360,
...     array_pr_anchor=[ActusDateTime(2024, 2, 15), ActusDateTime(2025, 1, 15)],
...     array_pr_cycle=["1M", "1M"],
...     array_pr_next=[1000.0, 2000.0],
...     array_increase_decrease=["INC", "DEC"]
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.05)
>>> contract = create_contract(attrs, rf_obs)
>>> result = contract.simulate()
jactus.contracts.lax.generate_array_schedule(anchors, cycles, end, filter_values=None, filter_target=None)[source]

Generate schedule from array of anchors and cycles.

If cycles is None or empty, each anchor date is treated as a single point event. If cycles is provided, each (anchor, cycle) pair generates a recurring sub-schedule bounded by the next anchor’s start date (segment boundaries).

Parameters:

  • anchors (list[ActusDateTime]) – Array of anchor dates (start dates for each sub-schedule)

  • cycles (list[str] | None) – Array of cycles (one per anchor), or None for point events

  • end (ActusDateTime) – End date (maturity date)

  • filter_values (list[str] | None) – Optional array of filter values (e.g., ARINCDEC)

  • filter_target (str | None) – Optional target value to filter for (e.g., “DEC”)

Returns:

Union of all sub-schedules, sorted and deduplicated

Return type:

list[ActusDateTime]

class jactus.contracts.lax.LAXPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for LAX contracts.

Extends LAM payoff functions with PI (Principal Increase) and PRF (Principal Redemption Amount Fixing) events.

ACTUS Reference:

ACTUS v1.1 Section 7.3 - LAX Payoff Functions

Events:

All LAM events (AD, IED, PR, MD, PP, PY, FP, PRD, TD, IP, IPCI, IPCB, RR, RRF, SC, CE) Plus: PI: Principal Increase (negative principal redemption) PRF: Principal Redemption Amount Fixing (update Prnxt)

Parameters:

  • contract_role (Any)

  • currency (str)

  • settlement_currency (str | None)

__init__(contract_role, currency, settlement_currency=None)[source]

Initialize LAX payoff function.

Parameters:

  • contract_role (Any) – Contract role (RPA or RPL)

  • currency (str) – Contract currency

  • settlement_currency (str | None) – Optional settlement currency

Return type:

None

calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for given event type.

Parameters:
Returns:

Payoff amount (JAX array)

Return type:

Array

class jactus.contracts.lax.LAXStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for LAX contracts.

Extends LAM state transitions with PI (Principal Increase) and PRF (Principal Redemption Amount Fixing) events.

ACTUS Reference:

ACTUS v1.1 Section 7.3 - LAX State Transition Functions

Key Differences from LAM:

  • PI events: Increase notional (opposite of PR)

  • PRF events: Fix Prnxt from array schedule

  • Array-based schedule generation

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer)[source]

Transition state for given event type.

Parameters:
Returns:

New contract state

Return type:

ContractState

class jactus.contracts.lax.ExoticLinearAmortizerContract(*args, **kwargs)[source]

Bases: BaseContract

LAX (Exotic Linear Amortizer) contract implementation.

LAX is the most complex amortizing contract, supporting flexible array schedules for principal redemption, interest payments, and rate resets.

ACTUS Reference:

ACTUS v1.1 Section 7.3 - LAX: Exotic Linear Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize LAX contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not LAX

initialize_state()[source]

Initialize LAX contract state.

Initializes all state variables including Prnxt and Ipcb (same as LAM).

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get LAX payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

LAX payoff function instance

Return type:

LAXPayoffFunction

get_state_transition_function(event_type)[source]

Get LAX state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

LAX state transition function instance

Return type:

LAXStateTransitionFunction

generate_event_schedule()[source]

Generate complete event schedule for LAX contract.

LAX uses array schedules to generate PR, PI, PRF, IP, RR, and RRF events.

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate LAX contract with array-aware prnxt injection.

Before each PR/PI event, updates state.prnxt from the array schedule so the correct principal amount is used without explicit PRF events.

Parameters:
Return type:

SimulationHistory

NAM - Negative Amortizer

Negative Amortizer (NAM) contract implementation.

This module implements the NAM contract type - an amortizing loan where principal can increase (negative amortization) when payments are less than accrued interest. NAM extends the LAM pattern with modified payoff and state transition functions.

ACTUS Reference:

ACTUS v1.1 Section 7.4 - NAM: Negative Amortizer

Key Features:

  • Negative amortization: Notional can increase if payment < interest

  • Modified PR payoff: Prnxt - accrued interest (can be negative)

  • Modified PR STF: Notional changes by net payment amount

  • IP schedule ends one period before PR schedule

  • Maturity calculation accounts for negative amortization effect

  • Same states as LAM: prnxt, ipcb

  • Same events as LAM

Negative Amortization:

When the scheduled principal payment (Prnxt) is less than the accrued interest, the shortfall is added to the notional principal:

  • If Prnxt > interest: Normal amortization (notional decreases)

  • If Prnxt < interest: Negative amortization (notional increases)

  • If Prnxt = interest: Interest-only payment (notional unchanged)

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.core import ActusDateTime, DayCountConvention
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="NAM-001",
...     contract_type=ContractType.NAM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2054, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=300000.0,
...     nominal_interest_rate=0.065,
...     day_count_convention=DayCountConvention.A360,
...     principal_redemption_cycle="1M",  # Monthly payments
...     next_principal_redemption_amount=800.0,  # Low payment → negative amort
...     interest_calculation_base="NT"
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.065)
>>> contract = create_contract(attrs, rf_obs)
>>> result = contract.simulate()
class jactus.contracts.nam.NAMPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for NAM contracts.

Implements all NAM payoff functions according to ACTUS specification. The key difference from LAM is the PR event payoff, which is net of accrued interest and can be negative.

ACTUS Reference:

ACTUS v1.1 Section 7.4 - NAM Payoff Functions

Events:

Same as LAM, but PR payoff modified: POF_PR = R(CNTRL) × Nsc × (Prnxt - Ipac - Y(Sd, t) × Ipnr × Ipcb)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer, child_contract_observer=None)[source]

Calculate payoff for NAM event via dict dispatch.

Parameters:
Returns:

JAX array containing the payoff amount

Return type:

Array

class jactus.contracts.nam.NAMStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for NAM contracts.

Implements all NAM state transitions according to ACTUS specification. The key difference from LAM is the PR event, which adjusts notional by the NET payment amount (payment - interest), allowing notional to increase.

ACTUS Reference:

ACTUS v1.1 Section 7.4 - NAM State Transition Functions

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer, child_contract_observer=None)[source]

Apply state transition for NAM event via dict dispatch.

Parameters:
Returns:

New contract state after event

Return type:

ContractState

class jactus.contracts.nam.NegativeAmortizerContract(*args, **kwargs)[source]

Bases: BaseContract

Negative Amortizer (NAM) contract.

Amortizing loan where principal can increase when payments are less than accrued interest (negative amortization). Extends LAM pattern with modified principal redemption handling.

ACTUS Reference:

ACTUS v1.1 Section 7.4 - NAM: Negative Amortizer

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize NAM contract.

Parameters:
Raises:
generate_event_schedule()[source]

Generate NAM event schedule per ACTUS specification.

Same as LAM schedule, with all event types supported.

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize NAM contract state.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get NAM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

NAM payoff function instance

Return type:

NAMPayoffFunction

get_state_transition_function(event_type)[source]

Get NAM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

NAM state transition function instance

Return type:

NAMStateTransitionFunction

ANN - Annuity

Annuity (ANN) contract implementation.

This module implements the ANN contract type - an amortizing loan with constant total payments (principal + interest). ANN extends NAM with automatic calculation of payment amounts using the ACTUS annuity formula.

ACTUS Reference:

ACTUS v1.1 Section 7.5 - ANN: Annuity

Key Features:

  • Constant total payment amount (principal + interest stays constant)

  • Automatic calculation of Prnxt using ACTUS annuity formula

  • Payment recalculation after rate resets (RR, RRF events)

  • Same negative amortization handling as NAM

  • Most common amortizing contract (standard mortgages)

Annuity Formula:
The ACTUS annuity formula calculates the constant payment amount:

A(s, T, n, a, r) = (n + a) / Σ[∏((1 + Y_i × r)^-1)]

Where:

s = start time T = maturity n = notional principal a = accrued interest r = nominal interest rate Y_i = year fraction for period i

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.core import ActusDateTime, DayCountConvention
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="MORTGAGE-001",
...     contract_type=ContractType.ANN,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2054, 1, 15, 0, 0, 0),  # 30 years
...     currency="USD",
...     notional_principal=300000.0,
...     nominal_interest_rate=0.065,
...     day_count_convention=DayCountConvention.A360,
...     principal_redemption_cycle="1M",  # Monthly payments
...     interest_calculation_base="NT"
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.065)
>>> contract = create_contract(attrs, rf_obs)
>>> result = contract.simulate()
>>> # Payment amount automatically calculated to fully amortize loan
class jactus.contracts.ann.ANNStateTransitionFunction[source]

Bases: BaseStateTransitionFunction

State transition function for ANN contracts.

Extends NAM state transitions with payment recalculation after rate changes. The key difference from NAM is that RR and RRF events recalculate Prnxt using the annuity formula to maintain constant total payments.

ACTUS Reference:

ACTUS v1.1 Section 7.5 - ANN State Transition Functions

__init__()[source]

Initialize ANN state transition function.

Return type:

None

transition_state(event_type, state, attributes, time, risk_factor_observer, child_contract_observer=None)[source]

Apply state transition for ANN event via dict dispatch.

ANN overrides RR/RRF to recalculate Prnxt using annuity formula. All other events delegate to NAM’s handlers.

Parameters:
Returns:

New contract state after event

Return type:

ContractState

class jactus.contracts.ann.AnnuityContract(*args, **kwargs)[source]

Bases: BaseContract

Annuity (ANN) contract.

Amortizing loan with constant total payment amount (principal + interest). The payment amount is automatically calculated using the ACTUS annuity formula to ensure the loan is fully amortized by maturity.

ACTUS Reference:

ACTUS v1.1 Section 7.5 - ANN: Annuity

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes

risk_factor_observer

Risk factor observer for market data

Example

See module docstring for usage example.

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize ANN contract.

Parameters:
Raises:
generate_event_schedule()[source]

Generate ANN event schedule per ACTUS specification.

Same as NAM/LAM schedule, with all event types supported.

Events before status_date are excluded.

Return type:

EventSchedule

initialize_state()[source]

Initialize ANN contract state.

Key Feature: If Prnxt is not provided in attributes, calculate it using the ACTUS annuity formula to ensure constant total payments.

When IED < SD (contract already existed), state is initialized as if STF_IED already ran.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get ANN payoff function.

ANN uses the same payoff function as NAM.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

ANN payoff function instance (same as NAM)

Return type:

NAMPayoffFunction

get_state_transition_function(event_type)[source]

Get ANN state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

ANN state transition function instance

Return type:

ANNStateTransitionFunction

CLM - Call Money

Call Money (CLM) contract implementation.

This module implements the CLM contract type - an on-demand loan where the maturity is not fixed at inception but determined by observed events (typically a call event from the lender or repayment from the borrower).

ACTUS Reference:

ACTUS v1.1 Section 7.6 - CLM: Call Money

Key Features:

  • No fixed maturity date at inception

  • Maturity determined from observed events

  • Single interest payment at termination

  • Optional periodic interest capitalization (IPCI)

  • Principal repaid when called or at observed maturity

  • Simpler than PAM - no regular IP schedule

Typical Use Cases:

  • Lines of credit

  • Overnight/call loans

  • Demand deposits

  • Flexible repayment loans

Example

>>> from jactus.contracts import create_contract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.core import ActusDateTime, DayCountConvention
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="LOC-001",
...     contract_type=ContractType.CLM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     # No maturity_date - determined dynamically
...     currency="USD",
...     notional_principal=50000.0,
...     nominal_interest_rate=0.08,
...     day_count_convention=DayCountConvention.A360,
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.08)
>>> contract = create_contract(attrs, rf_obs)
>>> result = contract.simulate()
class jactus.contracts.clm.CLMPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for CLM contracts.

CLM payoff functions are similar to PAM but simpler: - No regular IP events (only at maturity) - Maturity is dynamic (from observed events) - IPCI events capitalize interest periodically

ACTUS Reference:

ACTUS v1.1 Section 7.6 - CLM Payoff Functions

Events:

AD: Analysis Date (0.0) IED: Initial Exchange Date (disburse principal) MD: Maturity Date (return principal + accrued - dynamic) PR: Principal Repayment (from observer) FP: Fee Payment IP: Interest Payment (single event at maturity) IPCI: Interest Capitalization RR: Rate Reset RRF: Rate Reset Fixing CE: Credit Event

Parameters:
__init__(contract_role, currency, settlement_currency=None)[source]

Initialize CLM payoff function.

Parameters:

  • contract_role (ContractRole) – Contract role (RPA or RPL)

  • currency (str) – Contract currency

  • settlement_currency (str | None) – Optional settlement currency

Return type:

None

calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for given event type.

Parameters:
Returns:

Payoff amount (JAX array)

Return type:

Array

class jactus.contracts.clm.CLMStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for CLM contracts.

CLM state transitions are similar to PAM but without IPCB or Prnxt states.

ACTUS Reference:

ACTUS v1.1 Section 7.6 - CLM State Transition Functions

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer)[source]

Transition state for given event type.

Parameters:
Returns:

New contract state

Return type:

ContractState

class jactus.contracts.clm.CallMoneyContract(*args, **kwargs)[source]

Bases: BaseContract

CLM (Call Money) contract implementation.

CLM is an on-demand loan where maturity is determined by observed events rather than fixed at inception. Common for lines of credit and demand loans.

ACTUS Reference:

ACTUS v1.1 Section 7.6 - CLM: Call Money

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CLM contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not CLM

initialize_state()[source]

Initialize CLM contract state.

CLM state is simpler than LAM - no prnxt or ipcb states. When status_date >= IED, the IED event is skipped so state must be initialized from contract attributes directly.

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get CLM payoff function.

Parameters:

event_type (Any) – Type of event (not used, all events use same POF)

Returns:

CLM payoff function instance

Return type:

CLMPayoffFunction

get_state_transition_function(event_type)[source]

Get CLM state transition function.

Parameters:

event_type (Any) – Type of event (not used, all events use same STF)

Returns:

CLM state transition function instance

Return type:

CLMStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CLM contract with BDC-aware rate observation.

For SCP/SCF conventions, RR events use the original (unadjusted) schedule date for rate observation, not the BDC-shifted event time.

Parameters:
Return type:

SimulationHistory

generate_event_schedule()[source]

Generate complete event schedule for CLM contract.

CLM schedule is simpler than PAM: - No regular IP events (only at maturity if MD is set) - Optional IPCI events for interest capitalization - Maturity may be dynamic (from observed events)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

UMP - Undefined Maturity Profile

Undefined Maturity Profile (UMP) contract implementation.

This module implements the UMP contract type from the ACTUS specification. UMP represents a loan with uncertain principal repayment schedule where all principal changes come from observed events rather than a fixed schedule.

Key Features:

  • All PR/PI events from observer (no fixed schedule)

  • No IP schedule (only IPCI for capitalization)

  • Uncertain cashflows (all principal changes observed)

  • Maturity from last observed event

  • Same state variables as CLM (no prnxt or ipcb)

Example

>>> from jactus import UndefinedMaturityProfileContract, ContractAttributes
>>> from jactus.core.types import ContractType, ContractRole
>>> from jactus.core.datetime import ActusDateTime
>>>
>>> # Create UMP contract (line of credit with uncertain repayments)
>>> attrs = ContractAttributes(
...     contract_id="UMP-001",
...     contract_type=ContractType.UMP,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=100000.0,
...     nominal_interest_rate=0.06,
...     day_count_convention=DayCountConvention.A360,
...     # No maturity date - determined from observed events
...     # No IP cycle - only IPCI
...     ipci_cycle="1Q",  # Quarterly capitalization
... )
>>> contract = UndefinedMaturityProfileContract(attrs)
>>> # PR events will come from observer observations

References

ACTUS v1.1 Section 7.7 - Undefined Maturity Profile (UMP)

class jactus.contracts.ump.UMPPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for UMP contracts.

UMP payoffs are simpler than amortizing contracts: - IED: Disburse principal - PR: Return partial principal (from observer) - MD: Return remaining principal + accrued interest - IPCI: No payoff (capitalization is internal) - FP: Pay accrued fees - RR/RRF: No payoff

All principal repayment amounts come from observed events.

Parameters:
__init__(contract_role, currency, settlement_currency=None)[source]

Initialize base payoff function.

Parameters:

  • contract_role (ContractRole) – Contract role for sign adjustment

  • currency (str) – Contract currency (e.g., “USD”)

  • settlement_currency (str | None) – Settlement currency (None = same as contract)

Return type:

None

calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for given event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

jnp.ndarray

class jactus.contracts.ump.UMPStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for UMP contracts.

UMP state transitions are similar to CLM but with all PR from observer: - IED: Initialize notional and rate - PR: Reduce notional (amount from observer) - MD: Zero out all state variables - FP: Reset accrued fees - IPCI: Capitalize interest into notional - RR: Update interest rate - RRF: Fix interest rate

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer)[source]

Execute state transition for given event type.

Parameters:
Returns:

New contract state

Return type:

ContractState

class jactus.contracts.ump.UndefinedMaturityProfileContract(*args, **kwargs)[source]

Bases: BaseContract

Undefined Maturity Profile (UMP) contract.

UMP represents a loan with uncertain principal repayment schedule. All principal changes (PR/PI events) come from observed events rather than a fixed schedule. This is useful for modeling: - Lines of credit with uncertain drawdowns/repayments - Loans with irregular principal payments - Cashflow profiles determined by external events

Key Characteristics:

  • All PR events from observer (no fixed schedule)

  • No IP schedule (only IPCI for capitalization)

  • Uncertain cashflows (all principal from observations)

  • Maturity from last observed event

  • Simpler state than LAM (no prnxt or ipcb)

Event Types:

  • AD: Analysis Date

  • IED: Initial Exchange

  • PR: Principal Repayment (from observer)

  • PI: Principal Increase (from observer)

  • FP: Fee Payment

  • IPCI: Interest Capitalization

  • RR: Rate Reset

  • RRF: Rate Reset Fixing

  • CE: Credit Event

  • MD: Maturity

State Variables:

  • sd: Status Date

  • tmd: Terminal Maturity Date (from observed events)

  • nt: Notional

  • ipnr: Interest Rate

  • ipac: Accrued Interest

  • feac: Accrued Fees

  • nsc: Notional Scaling

  • isc: Interest Scaling

Example

>>> attrs = ContractAttributes(
...     contract_id="UMP-001",
...     contract_type=ContractType.UMP,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     currency="USD",
...     notional_principal=100000.0,
...     nominal_interest_rate=0.06,
...     ipci_cycle="1Q",  # Quarterly capitalization
... )
>>> contract = UndefinedMaturityProfileContract(attrs)

References

ACTUS v1.1 Section 7.7 - Undefined Maturity Profile

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize UMP contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes

  • risk_factor_observer (RiskFactorObserver) – Risk factor observer for rate updates

  • child_contract_observer (Any | None) – Optional child contract observer

Raises:

ValueError – If contract_type is not UMP

initialize_state()[source]

Initialize UMP state.

UMP state is simpler than LAM - no prnxt or ipcb states. Maturity is determined from observed events.

Return type:

ContractState

get_payoff_function(event_type)[source]

Get UMP payoff function.

Parameters:

event_type (Any) – Event type (unused - same POF for all)

Returns:

UMP payoff function instance

Return type:

UMPPayoffFunction

get_state_transition_function(event_type)[source]

Get UMP state transition function.

Parameters:

event_type (Any) – Event type (unused - same STF for all)

Returns:

UMP state transition function instance

Return type:

UMPStateTransitionFunction

generate_event_schedule()[source]

Generate UMP event schedule.

UMP schedule includes: - AD: Analysis date (if provided) - IED: Initial exchange - IPCI: Interest capitalization (from interest_payment_cycle) - RR: Rate reset (if RR cycle provided) - RRF: Rate reset fixing (if RRF cycle provided) - FP: Fee payment (if FP cycle provided) - TD: Termination (if termination_date is set) - MD: Maturity (from observed events or attributes)

Note: PR/PI events come from observer, not from schedule.

Returns:

Event schedule with all scheduled events

Return type:

EventSchedule

Non-Principal Contracts

CSH - Cash

Cash (CSH) contract implementation.

This module implements the CSH (Cash) contract type, which represents a simple cash position. It is the simplest ACTUS contract with only Analysis Date (AD) events and minimal state.

ACTUS Reference:

ACTUS v1.1 Section 7.8 - CSH: Cash

Example

>>> from jactus.contracts.csh import CashContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="CASH-001",
...     contract_type=ContractType.CSH,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     notional_principal=10000.0
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=0.0)
>>> contract = CashContract(attributes=attrs, risk_factor_observer=rf_obs)
>>> result = contract.simulate()
class jactus.contracts.csh.CashPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for Cash (CSH) contracts.

CSH contracts have no actual cashflows - they only represent a position. All payoffs return 0.0.

ACTUS Reference:

ACTUS v1.1 Section 7.8 - CSH Payoff Functions

Example

>>> pof = CashPayoffFunction(contract_role=ContractRole.RPA, currency="USD")
>>> state = ContractState(...)
>>> payoff = pof(EventType.AD, state, attrs, time, rf_obs)
>>> print(payoff)  # 0.0
Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for CSH events.

CSH contracts have no cashflows - they only represent a position value. All events return 0.0 payoff.

Parameters:
Returns:

Always returns 0.0 (no cashflow)

Return type:

Array

ACTUS Reference:

POF_AD_CSH = 0.0

class jactus.contracts.csh.CashStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for Cash (CSH) contracts.

CSH contracts have minimal state transitions - only the status date is updated. Notional remains constant throughout the contract life.

ACTUS Reference:

ACTUS v1.1 Section 7.8 - CSH State Transition Functions

Example

>>> stf = CashStateTransitionFunction()
>>> state_post = stf(EventType.AD, state_pre, attrs, time, rf_obs)
>>> assert state_post.sd == time
>>> assert state_post.nt == state_pre.nt  # Notional unchanged
Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Transition CSH contract state.

CSH state transitions are minimal - only the status date is updated. The notional (nt) remains constant.

Parameters:
Returns:

Updated contract state with new status date

Return type:

ContractState

ACTUS Reference:
STF_AD_CSH:

sd_t = t nt_t = nt_t⁻ (unchanged)

class jactus.contracts.csh.CashContract(*args, **kwargs)[source]

Bases: BaseContract

Cash (CSH) contract implementation.

Represents a simple cash position with no cashflows. This is the simplest ACTUS contract type, used for cash accounts, settlement amounts, or as initial positions in portfolios.

Characteristics:

  • Only AD (Analysis Date) events

  • Minimal state: notional (nt) and status date (sd)

  • No cashflows (all payoffs = 0.0)

  • No interest accrual or fees

ACTUS Reference:

ACTUS v1.1 Section 7.8 - CSH: Cash

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract attributes including notional, currency, role

risk_factor_observer

Observer for market data (not used in CSH)

child_contract_observer

Observer for child contracts (not used in CSH)

rngs

Random number generators for JAX

Example

>>> from jactus.contracts.csh import CashContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>>
>>> attrs = ContractAttributes(
...     contract_id="CASH-001",
...     contract_type=ContractType.CSH,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     notional_principal=10000.0
... )
>>>
>>> contract = CashContract(
...     attributes=attrs,
...     risk_factor_observer=ConstantRiskFactorObserver(0.0)
... )
>>>
>>> # Simulate the contract
>>> result = contract.simulate()
>>> print(f"Events: {len(result.events)}")  # 1 (just AD)
>>> print(f"Total cashflow: {result.total_cashflow()}")  # 0.0
__init__(attributes, risk_factor_observer, child_contract_observer=None, *, rngs=None)[source]

Initialize Cash contract.

Parameters:
Raises:

ValueError – If validation fails

Validations:

  • contract_type must be ContractType.CSH

  • notional_principal (NT) must be defined

  • contract_role (CNTRL) must be defined

  • currency (CUR) must be defined

generate_event_schedule()[source]

Generate CSH event schedule.

CSH contracts only have Analysis Date (AD) events. A single AD event is created at the status date.

Returns:

EventSchedule with one AD event

Return type:

EventSchedule

ACTUS Reference:
CSH Contract Schedule:

  • AD event at status_date

  • No other events

initialize_state()[source]

Initialize CSH contract state.

CSH state is minimal: - nt: Notional with role sign applied: R(CNTRL) × NT - sd: Status date - All other states are zero or null

Returns:

Initial contract state

Return type:

ContractState

ACTUS Reference:
CSH State Initialization:

nt = R(CNTRL) × NT sd = status_date All other states = 0.0 or null

get_payoff_function(event_type)[source]

Get payoff function for CSH events.

Parameters:

event_type (Any) – Type of event (only AD for CSH)

Returns:

CashPayoffFunction instance

Return type:

CashPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CSH events.

Parameters:

event_type (Any) – Type of event (only AD for CSH)

Returns:

CashStateTransitionFunction instance

Return type:

CashStateTransitionFunction

STK - Stock

Stock (STK) contract implementation.

This module implements the STK contract type - an equity position with dividend payments. STK is a simple contract representing stock ownership with potential dividend income.

ACTUS Reference:

ACTUS v1.1 Section 7.9 - STK: Stock

Key Features:

  • Equity position value from market observation

  • Fixed or market-observed dividend payments

  • Purchase and termination events

  • Minimal state (only performance and status date)

  • 6 event types total

Example

>>> from jactus.contracts.stk import StockContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="STK-001",
...     contract_type=ContractType.STK,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     market_object_code="AAPL",  # Stock ticker
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=150.0)  # Stock price
>>> contract = StockContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.stk.StockPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for STK contracts.

Implements all 6 STK payoff functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.9 - STK Payoff Functions

Events:

AD: Analysis Date (0.0) PRD: Purchase Date (pay purchase price) TD: Termination Date (receive termination price) DV(fix): Fixed Dividend Payment DV: Market-Observed Dividend Payment CE: Credit Event (0.0)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for STK events.

Dispatches to specific payoff function based on event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

ACTUS Reference:

POF_[event]_STK functions from Section 7.9

class jactus.contracts.stk.StockStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for STK contracts.

Implements all 6 STK state transition functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.9 - STK State Transition Functions

Note

STK has minimal state - only status date (sd) and performance (prf). All events simply update the status date.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Transition STK contract state.

All STK events have the same state transition: update status date only.

Parameters:
Returns:

Updated contract state

Return type:

ContractState

ACTUS Reference:

STF_[event]_STK functions from Section 7.9

class jactus.contracts.stk.StockContract(*args, **kwargs)[source]

Bases: BaseContract

Stock (STK) contract implementation.

Represents an equity position with potential dividend payments. STK is one of the simplest ACTUS contracts, with minimal state and straightforward cashflow logic.

ACTUS Reference:

ACTUS v1.1 Section 7.9

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and parameters

risk_factor_observer

Observer for market prices and dividends

child_contract_observer

Observer for child contracts (optional)

Example

>>> attrs = ContractAttributes(
...     contract_id="STK-001",
...     contract_type=ContractType.STK,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     market_object_code="AAPL",
... )
>>> contract = StockContract(attrs, risk_obs)
>>> result = contract.simulate()
__init__(attributes, risk_factor_observer, child_contract_observer=None, rngs=None)[source]

Initialize STK contract.

Parameters:
Raises:

ValueError – If contract_type is not STK or required attributes missing

generate_event_schedule()[source]

Generate STK event schedule.

Generates events for stock contract: - AD: Analysis dates (if specified) - PRD: Purchase date (if specified) - TD: Termination date (if specified) - DV: Dividend events (if dividend schedule specified)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

ACTUS Reference:

STK Contract Schedule from Section 7.9

initialize_state()[source]

Initialize STK contract state.

STK has minimal state - only status date and performance.

ACTUS Reference:

STK State Initialization from Section 7.9

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for STK events.

Parameters:

event_type (Any)

Return type:

StockPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for STK events.

Parameters:

event_type (Any)

Return type:

StockStateTransitionFunction

COM - Commodity

Commodity (COM) contract implementation.

This module implements the COM contract type - a commodity position with price exposure. COM is a simple contract representing commodity ownership (e.g., gold, oil, wheat).

ACTUS Reference:

ACTUS v1.1 Section 7.10 - COM: Commodity

Key Features:

  • Commodity position value from market observation

  • Purchase and termination events

  • Minimal state (only performance and status date)

  • 4 event types: AD, PRD, TD, CE

Example

>>> from jactus.contracts.com import CommodityContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="COM-001",
...     contract_type=ContractType.COM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     price_at_purchase_date=7500.0,  # Total purchase price
...     price_at_termination_date=8200.0,  # Total sale price
... )
>>>
>>> rf_obs = ConstantRiskFactorObserver(constant_value=80.0)
>>> contract = CommodityContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.com.CommodityPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for COM contracts.

Implements all COM payoff functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.10 - COM Payoff Functions

Events:

AD: Analysis Date (0.0) PRD: Purchase Date (pay purchase price) TD: Termination Date (receive termination price) CE: Credit Event (0.0)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for COM events.

Dispatches to specific payoff function based on event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

ACTUS Reference:

POF_[event]_COM functions from Section 7.10

class jactus.contracts.com.CommodityStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for COM contracts.

Implements all COM state transition functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.10 - COM State Transition Functions

Note

COM has minimal state - only status date (sd) and performance (prf). All events simply update the status date.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Transition COM contract state.

All COM events have the same state transition: update status date only.

Parameters:
Returns:

Updated contract state

Return type:

ContractState

ACTUS Reference:

STF_[event]_COM functions from Section 7.10

class jactus.contracts.com.CommodityContract(*args, **kwargs)[source]

Bases: BaseContract

Commodity (COM) contract implementation.

Represents a commodity position with price exposure. COM is one of the simplest ACTUS contracts, similar to STK but for physical or financial commodities (gold, oil, wheat, etc.).

ACTUS Reference:

ACTUS v1.1 Section 7.10

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and parameters

risk_factor_observer

Observer for market prices

child_contract_observer

Observer for child contracts (optional)

Example

>>> attrs = ContractAttributes(
...     contract_id="COM-001",
...     contract_type=ContractType.COM,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     currency="USD",
...     price_at_purchase_date=7500.0,
...     price_at_termination_date=8200.0,
... )
>>> contract = CommodityContract(attrs, risk_obs)
>>> result = contract.simulate()
__init__(attributes, risk_factor_observer, child_contract_observer=None, rngs=None)[source]

Initialize COM contract.

Parameters:
Raises:

ValueError – If contract_type is not COM or required attributes missing

generate_event_schedule()[source]

Generate COM event schedule.

Generates events for commodity contract: - AD: Analysis dates (if specified) - PRD: Purchase date (if specified) - TD: Termination date (if specified)

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

ACTUS Reference:

COM Contract Schedule from Section 7.10

initialize_state()[source]

Initialize COM contract state.

COM has minimal state - only status date and performance.

ACTUS Reference:

COM State Initialization from Section 7.10

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for COM events.

Parameters:

event_type (Any)

Return type:

CommodityPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for COM events.

Parameters:

event_type (Any)

Return type:

CommodityStateTransitionFunction

Derivative Contracts

FXOUT - Foreign Exchange Outright

Foreign Exchange Outright (FXOUT) contract implementation.

This module implements the FXOUT contract type - an FX forward or spot contract with settlement in two currencies. FXOUT represents the exchange of two currency amounts at a future date (or spot).

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT: Foreign Exchange Outright

Key Features:

  • Dual currency settlement (CUR and CUR2)

  • Two settlement modes: delivery (net) or dual (gross)

  • FX rate observation at settlement

  • No interest payments or principal amortization

  • 7 event types total

Settlement Modes:

  • Delivery (DS=’D’): Net settlement in a single currency

  • Dual (DS=’S’): Two separate payments, one in each currency

Example

>>> from jactus.contracts.fxout import FXOutrightContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> # EUR/USD forward: buy 100,000 EUR, sell 110,000 USD
>>> attrs = ContractAttributes(
...     contract_id="FXFWD-001",
...     contract_type=ContractType.FXOUT,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2024, 7, 1, 0, 0, 0),
...     currency="EUR",  # First currency
...     currency_2="USD",  # Second currency
...     notional_principal=100000.0,  # EUR amount
...     notional_principal_2=110000.0,  # USD amount (forward rate = 1.10)
...     delivery_settlement="D",  # Net settlement
... )
>>>
>>> # FX rate observer (returns USD/EUR rate)
>>> rf_obs = ConstantRiskFactorObserver(constant_value=1.12)
>>> contract = FXOutrightContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.fxout.FXOutrightPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for FXOUT contracts.

Implements all 7 FXOUT payoff functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT Payoff Functions

Events:

AD: Analysis Date (0.0) PRD: Purchase Date (pay purchase price) TD: Termination Date (receive termination price) STD: Settlement Date (net settlement, delivery mode) STD(1): Settlement Date - first currency (dual mode) STD(2): Settlement Date - second currency (dual mode) CE: Credit Event (0.0)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for FXOUT events.

Dispatches to specific payoff function based on event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

ACTUS Reference:

POF_[event]_FXOUT functions from Section 7.11

class jactus.contracts.fxout.FXOutrightStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for FXOUT contracts.

Implements all FXOUT state transition functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT State Transition Functions

State Variables:

tmd: Maturity date prf: Contract performance (DF, DL, DQ, PF) sd: Status date

Events:

AD: Update status date PRD: Update status date TD: Update status date STD: Update status date CE: Update performance and status date

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate state transition for FXOUT events.

Dispatches to specific state transition function based on event type.

Parameters:
Returns:

New contract state

Return type:

ContractState

ACTUS Reference:

STF_[event]_FXOUT functions from Section 7.11

class jactus.contracts.fxout.FXOutrightContract(*args, **kwargs)[source]

Bases: BaseContract

Foreign Exchange Outright (FXOUT) contract.

Represents an FX forward or spot contract with settlement in two currencies.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT: Foreign Exchange Outright

Key Attributes:

currency (CUR): First currency (e.g., “EUR”) currency_2 (CUR2): Second currency (e.g., “USD”) notional_principal (NT): Amount in first currency notional_principal_2 (NT2): Amount in second currency delivery_settlement (DS): ‘D’ (delivery/net) or ‘S’ (dual/gross) maturity_date (MD) or settlement_date (STD): Settlement date

Settlement Modes:
Delivery (DS=’D’):

  • Single STD event with net settlement

  • Payoff = NT - (FX_rate × NT2)

  • Settled in first currency

Dual (DS=’S’):

  • Two STD events: STD(1) and STD(2)

  • STD(1): Receive NT in first currency

  • STD(2): Pay NT2 in second currency

  • Full principal exchange

State Variables:

md: Maturity/settlement date prf: Contract performance sd: Status date

Example

EUR/USD forward contract: - Buy 100,000 EUR - Sell 110,000 USD - Forward rate = 1.10 (agreed) - At maturity, observe market rate - If market rate = 1.12, profit = 100,000 - (100,000 × 110,000/112,000) ≈ 1,786 EUR

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize FXOUT contract.

Parameters:
Raises:

ValueError – If contract type is not FXOUT or required attributes missing

generate_event_schedule()[source]

Generate event schedule for FXOUT contract.

Events depend on delivery/settlement mode and settlement period: - DS=’S’ with settlement period → single STD at maturity + SP (net settlement) - Otherwise → MD events at maturity (gross exchange, two currency legs) - Early termination (TD < maturity) → suppress MD/STD events

Returns:

EventSchedule with contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state.

ACTUS Reference:

ACTUS v1.1 Section 7.11 - FXOUT State Variables Initialization

State Variables:

tmd: MD if STD = ∅, else STD prf: PRF (contract performance) sd: Status date nt, ipnr, ipac, feac, nsc, isc: Zero/default values (not used by FXOUT)

Returns:

Initial contract state

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for FXOUT contract.

Parameters:

event_type (Any) – Event type (not used for FXOUT, same function for all events)

Returns:

FXOutrightPayoffFunction instance

Return type:

BasePayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for FXOUT contract.

Parameters:

event_type (Any) – Event type (not used for FXOUT, same function for all events)

Returns:

FXOutrightStateTransitionFunction instance

Return type:

BaseStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate FXOUT contract with dual-currency MD and net STD handling.

Overrides base simulate() to compute: - Per-leg payoffs for gross settlement (MD events) - Net settlement payoff for cash settlement (STD events)

Parameters:
Return type:

SimulationHistory

OPTNS - Options

Option Contract (OPTNS) implementation.

This module implements the OPTNS contract type - a vanilla option contract with call, put, and collar options supporting European, American, and Bermudan exercise styles.

ACTUS Reference:

ACTUS v1.1 Section 7.15 - OPTNS: Option

Key Features:

  • Option types: Call (‘C’), Put (‘P’), Collar (‘CP’)

  • Exercise types: European (‘E’), American (‘A’), Bermudan (‘B’)

  • Underlier reference via contract_structure (CTST)

  • Exercise decision logic based on intrinsic value

  • Settlement after exercise

  • Premium payment at purchase

Exercise Mechanics:

  • European: Exercise only at maturity date

  • American: Exercise anytime before expiration

  • Bermudan: Exercise on specific dates only

Settlement:

  • Exercise Date (XD): Calculate exercise amount Xa

  • Settlement Date (STD): Receive Xa (after settlement period)

Example

>>> from jactus.contracts.optns import OptionContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> # European call option on stock with $100 strike
>>> attrs = ContractAttributes(
...     contract_id="OPT-CALL-001",
...     contract_type=ContractType.OPTNS,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2024, 12, 31, 0, 0, 0),
...     currency="USD",
...     notional_principal=100.0,  # Number of shares
...     option_type="C",  # Call option
...     option_strike_1=100.0,  # Strike price
...     option_exercise_type="E",  # European
...     price_at_purchase_date=5.0,  # Premium per share
...     contract_structure="AAPL",  # Underlier (stock ticker)
... )
>>>
>>> # Risk factor observer for stock price
>>> rf_obs = ConstantRiskFactorObserver(constant_value=110.0)  # Stock at $110
>>> contract = OptionContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.optns.OptionPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for OPTNS contracts.

Implements all OPTNS payoff functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.15 - OPTNS Payoff Functions

Events:

AD: Analysis Date (0.0) PRD: Purchase Date (pay premium) TD: Termination Date (receive termination price) MD: Maturity Date (automatic exercise if in-the-money) XD: Exercise Date (exercise decision) STD: Settlement Date (receive exercise amount) CE: Credit Event (contract default)

State Variables Used:

xa: Exercise amount (calculated at XD) prf: Contract performance (default status)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for OPTNS events.

Dispatches to specific payoff function based on event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

class jactus.contracts.optns.OptionStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for OPTNS contracts.

Handles state transitions for option contracts, including: - Exercise decision logic - Exercise amount calculation - State updates after settlement

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for a given event.

Parameters:
Returns:

Updated contract state (after event)

Return type:

ContractState

class jactus.contracts.optns.OptionContract(*args, **kwargs)[source]

Bases: BaseContract

Option Contract (OPTNS) implementation.

Represents a vanilla option contract with call, put, or collar payoffs. Supports European, American, and Bermudan exercise styles.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

option_type

‘C’ (call), ‘P’ (put), ‘CP’ (collar)

Type:

OPTP

option_strike_1

Primary strike price

Type:

OPS1

option_strike_2

Secondary strike (collar only)

Type:

OPS2

option_exercise_type

‘E’ (European), ‘A’ (American), ‘B’ (Bermudan)

Type:

OPXT

contract_structure

Underlier reference

Type:

CTST

notional_principal

Number of units (e.g., shares)

Type:

NT

price_at_purchase_date

Premium per unit

Type:

PPRD

maturity_date

Option expiration date

Type:

MD

settlement_period

Period from exercise to settlement

Type:

STPD

Example

>>> # European call option
>>> attrs = ContractAttributes(
...     contract_type=ContractType.OPTNS,
...     option_type="C",
...     option_strike_1=100.0,
...     option_exercise_type="E",
...     contract_structure="AAPL",
...     ...
... )
>>> contract = OptionContract(attrs, rf_obs)
>>> events = contract.generate_event_schedule()
__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize OPTNS contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate event schedule for OPTNS contract.

Events depend on exercise type: - European: AD (optional), PRD, MD, XD, STD - American: AD (optional), PRD, XD (multiple), MD, STD - Bermudan: AD (optional), PRD, XD (specific dates), MD, STD - Pre-exercised (exercise_date + exercise_amount set): STD only

Returns:

Event schedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial contract state with xa set to exercise_amount if pre-exercised

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for OPTNS contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

OptionPayoffFunction instance

Return type:

OptionPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for OPTNS contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

OptionStateTransitionFunction instance

Return type:

OptionStateTransitionFunction

FUTUR - Futures

Future Contract (FUTUR) implementation.

This module implements the FUTUR contract type - a futures contract with linear payoff based on the difference between spot and futures price.

ACTUS Reference:

ACTUS v1.1 Section 7.16 - FUTUR: Future

Key Features:

  • Linear payoff: Xa = S_t - PFUT (can be positive or negative)

  • No premium payment (unlike options)

  • Always settles at maturity

  • Mark-to-market settlement

  • Underlier reference via contract_structure (CTST)

Differences from Options:

  • No exercise decision logic (always settles)

  • Linear payoff (not max-based like options)

  • No option premium

  • Settlement amount can be negative

Example

>>> from jactus.contracts.futur import FutureContract
>>> from jactus.core import ContractAttributes, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> # Gold futures contract at $1800/oz
>>> attrs = ContractAttributes(
...     contract_id="FUT-GOLD-001",
...     contract_type=ContractType.FUTUR,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2024, 12, 31, 0, 0, 0),
...     currency="USD",
...     notional_principal=100.0,  # 100 ounces
...     future_price=1800.0,  # Agreed futures price
...     contract_structure="GC",  # Underlier (gold)
... )
>>>
>>> # Risk factor observer for gold price
>>> rf_obs = ConstantRiskFactorObserver(constant_value=1850.0)  # Spot at $1850
>>> contract = FutureContract(
...     attributes=attrs,
...     risk_factor_observer=rf_obs
... )
>>> result = contract.simulate()
class jactus.contracts.futur.FuturePayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for FUTUR contracts.

Implements all FUTUR payoff functions according to ACTUS specification.

ACTUS Reference:

ACTUS v1.1 Section 7.16 - FUTUR Payoff Functions

Events:

AD: Analysis Date (0.0) PRD: Purchase Date (zero - no premium) TD: Termination Date (receive termination price) MD: Maturity Date (settlement amount calculated) STD: Settlement Date (receive settlement amount Xa) CE: Credit Event (contract default)

State Variables Used:

xa: Settlement amount (calculated at MD) = S_t - PFUT prf: Contract performance (default status)

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for FUTUR events.

Dispatches to specific payoff function based on event type.

Parameters:
Returns:

Payoff amount as JAX array

Return type:

Array

class jactus.contracts.futur.FutureStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for FUTUR contracts.

Handles state transitions for futures contracts, including: - Settlement amount calculation (linear payoff) - State updates after settlement

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for a given event.

Parameters:
Returns:

Updated contract state (after event)

Return type:

ContractState

class jactus.contracts.futur.FutureContract(*args, **kwargs)[source]

Bases: BaseContract

Future Contract (FUTUR) implementation.

Represents a futures contract with linear payoff based on spot vs futures price.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

future_price

Agreed futures price

Type:

PFUT

contract_structure

Underlier reference

Type:

CTST

notional_principal

Number of units (e.g., contracts)

Type:

NT

maturity_date

Futures expiration date

Type:

MD

Key Differences from Options:

  • Linear payoff (not max-based)

  • No premium payment

  • Settlement amount can be negative

  • Always settles at maturity (no exercise decision)

Example

>>> # S&P 500 futures contract
>>> attrs = ContractAttributes(
...     contract_type=ContractType.FUTUR,
...     future_price=4500.0,
...     contract_structure="SPX",
...     notional_principal=1.0,
...     maturity_date=ActusDateTime(2024, 12, 31, 0, 0, 0),
...     ...
... )
>>> contract = FutureContract(attrs, rf_obs)
>>> events = contract.generate_event_schedule()
__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize FUTUR contract.

Parameters:
Raises:

ValueError – If validation fails

generate_event_schedule()[source]

Generate event schedule for FUTUR contract.

Events: - AD (optional): Analysis dates - PRD (optional): Purchase date (zero payment) - TD (optional): Termination date - MD: Maturity date (settlement amount calculated) - XD: Exercise date at maturity - STD: Settlement date (receive settlement amount) - Pre-exercised (exercise_date + exercise_amount set): STD only

Returns:

Event schedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial contract state with xa set to exercise_amount if pre-exercised

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for FUTUR contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

FuturePayoffFunction instance

Return type:

FuturePayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for FUTUR contract.

Parameters:

event_type (Any) – The event type (for compatibility with BaseContract)

Returns:

FutureStateTransitionFunction instance

Return type:

FutureStateTransitionFunction

SWPPV - Plain Vanilla Swap

Plain Vanilla Interest Rate Swap (SWPPV) contract implementation.

This module implements a plain vanilla interest rate swap where one party pays a fixed rate and receives a floating rate (or vice versa). The swap exchanges interest payments on a notional amount without exchanging the principal.

Key Features:

  • Fixed leg: Uses nominal_interest_rate (IPNR)

  • Floating leg: Uses nominal_interest_rate_2 (IPNR2) with rate resets

  • Separate accrual tracking (ipac1 for fixed, ipac2 for floating)

  • Net or gross settlement modes

  • No notional exchange at inception or maturity

Example

>>> from jactus.contracts import PlainVanillaSwapContract
>>> from jactus.core import ContractAttributes, ActusDateTime
>>> from jactus.observers import ConstantRiskFactorObserver
>>>
>>> # Receive fixed, pay floating
>>> attrs = ContractAttributes(
...     contract_id="SWAP-001",
...     contract_type=ContractType.SWPPV,
...     contract_role=ContractRole.RPA,  # Receive fixed
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     contract_deal_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     initial_exchange_date=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 15, 0, 0, 0),
...     notional_principal=1000000.0,
...     nominal_interest_rate=0.05,  # Fixed leg: 5%
...     nominal_interest_rate_2=0.03,  # Floating leg initial: 3%
...     interest_payment_cycle="P6M",  # Semi-annual
...     rate_reset_cycle="P3M",  # Quarterly resets
...     delivery_settlement="D",  # Net settlement
... )
>>> rf_obs = ConstantRiskFactorObserver(0.04)
>>> swap = PlainVanillaSwapContract(attrs, rf_obs)
>>> cashflows = swap.simulate(rf_obs)
class jactus.contracts.swppv.PlainVanillaSwapPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for SWPPV contracts.

Calculates cashflows for fixed and floating leg interest payments.

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Dispatcher for payoff functions.

Parameters:
Return type:

Array

class jactus.contracts.swppv.PlainVanillaSwapStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for SWPPV contracts.

Manages accrual tracking for fixed and floating legs.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for swap events.

Parameters:
Return type:

ContractState

class jactus.contracts.swppv.PlainVanillaSwapContract(*args, **kwargs)[source]

Bases: BaseContract

Plain Vanilla Interest Rate Swap (SWPPV) contract.

Swaps fixed and floating interest rate payments on a notional amount. No exchange of principal occurs.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize SWPPV contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for SWPPV contract.

Returns:

EventSchedule with all contract events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for SWPPV contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

PlainVanillaSwapPayoffFunction instance

Return type:

PlainVanillaSwapPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for SWPPV contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

PlainVanillaSwapStateTransitionFunction instance

Return type:

PlainVanillaSwapStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate SWPPV contract.

Overrides base to filter out events before purchaseDate and after terminationDate. The full event schedule is processed for state computation, but only visible events are returned.

Parameters:
Return type:

SimulationHistory

SWAPS - Generic Swap

Generic Swap (SWAPS) contract implementation.

This module implements a generic swap contract where two legs are represented as explicit child contracts. This is the most flexible swap implementation, supporting any combination of contract types for the legs.

Key Features:

  • Two explicit child contract legs (FirstLeg, SecondLeg)

  • Event merging for congruent events (net settlement)

  • Leg role assignment based on parent role

  • State aggregation from both legs

  • Supports any contract type for legs (PAM, LAM, ANN, etc.)

Example

>>> from jactus.contracts import GenericSwapContract
>>> from jactus.core import ContractAttributes, ActusDateTime
>>> from jactus.observers import ConstantRiskFactorObserver, MockChildContractObserver
>>>
>>> # Create swap with PAM legs
>>> attrs = ContractAttributes(
...     contract_id="SWAP-001",
...     contract_type=ContractType.SWAPS,
...     contract_role=ContractRole.RFL,  # Receive first leg
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 1, 0, 0, 0),
...     delivery_settlement="D",  # Net settlement
...     contract_structure='{"FirstLeg": "LEG1-ID", "SecondLeg": "LEG2-ID"}',
... )
>>> rf_obs = ConstantRiskFactorObserver(0.03)
>>> child_obs = MockChildContractObserver()
>>> swap = GenericSwapContract(attrs, rf_obs, child_obs)
>>> cashflows = swap.simulate(rf_obs, child_obs)

References

ACTUS Technical Specification v1.1, Section 7.13

jactus.contracts.swaps.determine_leg_roles(parent_role)[source]

Determine leg roles based on parent contract role.

ACTUS Rule:

  • If parent CNTRL=’RFL’ (Receive First Leg): FirstLeg=RPA, SecondLeg=RPL

  • Otherwise: FirstLeg=RPL, SecondLeg=RPA

Parameters:

parent_role (ContractRole) – Parent contract role

Returns:

Tuple of (first_leg_role, second_leg_role)

Return type:

tuple[ContractRole, ContractRole]

Example

>>> determine_leg_roles(ContractRole.RFL)
(<ContractRole.RPA: 'RPA'>, <ContractRole.RPL: 'RPL'>)
>>> determine_leg_roles(ContractRole.PFL)
(<ContractRole.RPL: 'RPL'>, <ContractRole.RPA: 'RPA'>)
jactus.contracts.swaps.merge_congruent_events(event1, event2)[source]

Merge two congruent events (same time and type) into net event.

Congruent events have:

  • Same event time

  • Same event type

  • Compatible for netting (IED, IP, PR, MD)

Formula: f(z) = f(x) + f(y) where τ=t=s

Parameters:
Returns:

Merged event with summed payoffs and aggregated state

Return type:

ContractEvent

class jactus.contracts.swaps.GenericSwapPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for SWAPS contracts.

SWAPS payoffs are derived from child contract events.

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for generic swap events.

For SWAPS, payoffs come from child contract events which are already calculated and merged. This function returns zero as the actual payoffs are in the event schedule from children.

Parameters:
Returns:

Zero payoff (actual payoffs from child events)

Return type:

Array

class jactus.contracts.swaps.GenericSwapStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for SWAPS contracts.

SWAPS state is aggregated from child contract states.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for swap events.

For SWAPS, state transitions come from child contracts. We aggregate state variables from both legs.

Parameters:
Returns:

Updated contract state (aggregated from legs)

Return type:

ContractState

class jactus.contracts.swaps.GenericSwapContract(*args, **kwargs)[source]

Bases: BaseContract

Generic Swap (SWAPS) contract.

A swap with two explicit child contract legs. Supports any contract types for the legs (PAM, LAM, ANN, etc.) and provides flexible event merging and state aggregation.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for child contract data (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize SWAPS contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for SWAPS contract.

The schedule is created by: 1. Querying events from both child legs 2. Merging congruent events if DS=’D’ (net settlement) 3. Keeping all events separate if DS=’S’ (gross settlement)

Returns:

EventSchedule with merged or separate leg events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State is aggregated from both child leg states.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for SWAPS contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

GenericSwapPayoffFunction instance

Return type:

GenericSwapPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for SWAPS contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

GenericSwapStateTransitionFunction instance

Return type:

GenericSwapStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate SWAPS contract by passing through child contract events.

For SWAPS, event payoffs and states come directly from child contract simulations. The schedule events already contain pre-computed data, so we pass them through instead of recalculating via POF/STF.

Parameters:
Return type:

SimulationHistory

CAPFL - Cap/Floor

Cap-Floor (CAPFL) contract implementation.

This module implements interest rate cap and floor contracts. A cap/floor protects against interest rate movements:

  • Cap: pays max(0, floating_rate - cap_rate) * NT * YF

  • Floor: pays max(0, floor_rate - floating_rate) * NT * YF

  • Collar: both cap and floor

The CAPFL wraps an underlier (typically PAM or SWPPV) and generates events on the underlier’s IP/RR schedule.

Example

>>> from jactus.contracts import CapFloorContract
>>> from jactus.core import ContractAttributes, ActusDateTime, ContractType, ContractRole
>>> from jactus.observers import ConstantRiskFactorObserver, MockChildContractObserver
>>>
>>> attrs = ContractAttributes(
...     contract_id="CAP-001",
...     contract_type=ContractType.CAPFL,
...     contract_role=ContractRole.BUY,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 1, 0, 0, 0),
...     rate_reset_cap=0.06,
...     contract_structure='{"Underlying": "SWAP-001"}',
... )
>>> rf_obs = ConstantRiskFactorObserver(0.03)
>>> child_obs = MockChildContractObserver()
>>> cap = CapFloorContract(attrs, rf_obs, child_obs)

References

ACTUS Technical Specification v1.1, Section 7.14

class jactus.contracts.capfl.CapFloorPayoffFunction(contract_role=None, currency=None, settlement_currency=None, cap_rate=None, floor_rate=None, notional=0.0, day_count_convention=DayCountConvention.A365)[source]

Bases: BasePayoffFunction

Payoff function for CAPFL contracts.

Computes cap/floor differential payoffs at IP events.

Parameters:
__init__(contract_role=None, currency=None, settlement_currency=None, cap_rate=None, floor_rate=None, notional=0.0, day_count_convention=DayCountConvention.A365)[source]

Initialize base payoff function.

Parameters:

  • contract_role (ContractRole | None) – Contract role for sign adjustment

  • currency (str | None) – Contract currency (e.g., “USD”)

  • settlement_currency (str | None) – Settlement currency (None = same as contract)

  • cap_rate (float | None)

  • floor_rate (float | None)

  • notional (float)

  • day_count_convention (DayCountConvention)

calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate cap/floor payoff.

For IP events, computes the cap/floor differential:

Cap: max(0, floating_rate - cap_rate) * NT * YF Floor: max(0, floor_rate - floating_rate) * NT * YF

Parameters:
Return type:

Array

class jactus.contracts.capfl.CapFloorStateTransitionFunction(dcc=DayCountConvention.A365)[source]

Bases: BaseStateTransitionFunction

State transition function for CAPFL contracts.

Tracks the floating rate through RR events and advances sd at IP events.

Parameters:

dcc (DayCountConvention)

__init__(dcc=DayCountConvention.A365)[source]

Initialize state transition function.

Parameters:

  • day_count_convention – Day count convention for time calculations (if None, uses contract’s DCC from attributes)

  • dcc (DayCountConvention)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Implement contract-specific state transition logic.

This method should be implemented by subclasses to define how the contract state transitions for specific event types.

Parameters:
Returns:

Post-event contract state

Return type:

ContractState

Note

This method should typically: 1. Copy the pre-event state 2. Update state variables based on event type 3. Return the new state

Example

>>> def transition_state(self, event_type, state_pre, attributes, time, observer):
...     # Create post-event state by copying pre-event state
...     state_post = state_pre
...
...     # Update state based on event type
...     if event_type == EventType.IP:
...         # Interest payment: reset accrued interest
...         state_post = state_post.replace(ipac=jnp.array(0.0))
...
...     return state_post
class jactus.contracts.capfl.CapFloorContract(*args, **kwargs)[source]

Bases: BaseContract

Cap-Floor (CAPFL) contract.

An interest rate cap or floor that pays the differential when the floating rate breaches the cap or floor rate.

The contract either: 1. References an underlier via contract_structure (child observer mode) 2. Contains embedded underlier terms for standalone operation

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize base contract.

Parameters:

  • attributes (ContractAttributes) – Contract attributes (terms and conditions)

  • risk_factor_observer (RiskFactorObserver) – Observer for accessing market risk factors

  • child_contract_observer (ChildContractObserver | None) – Optional observer for child contracts

  • rngs – Optional Flax RNG state for stochastic contracts

Example

>>> from jactus.observers import ConstantRiskFactorObserver
>>> attrs = ContractAttributes(...)
>>> risk_obs = ConstantRiskFactorObserver(1.0)
>>> contract = MyContract(attrs, risk_obs)
generate_event_schedule()[source]

Generate event schedule for CAPFL contract.

If underlier terms are embedded, generates IP and RR schedules directly from those terms. Otherwise, queries the child observer.

Return type:

EventSchedule

initialize_state()[source]

Initialize CAPFL contract state.

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for a specific event type.

Returns the appropriate payoff function (POF) for calculating the cashflow generated by the given event type.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

PayoffFunction for the event type

Return type:

CapFloorPayoffFunction

Example

>>> def get_payoff_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentPayoff()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionPayoff()
...     else:
...         return ZeroPayoff()

References

ACTUS v1.1 Section 2.7 - Payoff Functions

get_state_transition_function(event_type)[source]

Get state transition function for a specific event type.

Returns the appropriate state transition function (STF) for updating contract state when the given event occurs.

Parameters:

event_type (Any) – Event type (e.g., EventType.IP, EventType.PR)

Returns:

StateTransitionFunction for the event type

Return type:

CapFloorStateTransitionFunction

Example

>>> def get_state_transition_function(self, event_type):
...     if event_type == EventType.IP:
...         return InterestPaymentSTF()
...     elif event_type == EventType.PR:
...         return PrincipalRedemptionSTF()
...     else:
...         return IdentitySTF()

References

ACTUS v1.1 Section 2.8 - State Transition Functions

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CAPFL contract.

RR events are used internally for rate tracking but filtered from the output since CAPFL only exposes IP events externally.

Parameters:
Return type:

SimulationHistory

CEG - Credit Enhancement Guarantee

Credit Enhancement Guarantee (CEG) contract implementation.

This module implements credit enhancement guarantee contracts that cover losses on covered contracts when credit events occur. Similar to credit default swaps, CEG contracts pay out when the performance of covered contracts deteriorates to a specified credit event type.

Key Features:

  • Covers losses on one or more contracts

  • Credit event triggers payout

  • Coverage amount calculated from covered contracts

  • Guarantee fees paid periodically

  • Multiple coverage extent modes (NO, NI, MV)

Example

>>> from jactus.contracts import CreditEnhancementGuaranteeContract
>>> from jactus.core import ContractAttributes, ActusDateTime
>>> from jactus.observers import ConstantRiskFactorObserver, MockChildContractObserver
>>>
>>> # Create credit guarantee covering a loan
>>> attrs = ContractAttributes(
...     contract_id="CEG-001",
...     contract_type=ContractType.CEG,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 1, 0, 0, 0),
...     coverage=0.8,  # 80% coverage
...     credit_event_type=ContractPerformance.DL,  # Default
...     credit_enhancement_guarantee_extent="NO",  # Notional only
...     contract_structure='{"CoveredContract": "LOAN-001"}',
...     fee_rate=0.01,  # 1% annual fee
...     fee_payment_cycle="P1Y",
... )
>>> rf_obs = ConstantRiskFactorObserver(0.03)
>>> child_obs = MockChildContractObserver()
>>> ceg = CreditEnhancementGuaranteeContract(attrs, rf_obs, child_obs)
>>> cashflows = ceg.simulate(rf_obs, child_obs)

References

ACTUS Technical Specification v1.1, Section 7.17

class jactus.contracts.ceg.CEGPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for CEG contracts.

CEG payoffs include guarantee fees and credit event payouts.

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for credit guarantee events.

Parameters:
Returns:

Payoff amount (fees or credit event payout)

Return type:

Array

class jactus.contracts.ceg.CEGStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for CEG contracts.

CEG state tracks coverage amount, fee accrual, and exercise status.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for guarantee events.

Parameters:
Returns:

Updated contract state

Return type:

ContractState

class jactus.contracts.ceg.CreditEnhancementGuaranteeContract(*args, **kwargs)[source]

Bases: BaseContract

Credit Enhancement Guarantee (CEG) contract.

A guarantee contract that pays out when covered contracts experience credit events. The payout covers a specified percentage of the covered amount, calculated based on the coverage extent mode (notional only, notional plus interest, or market value).

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for covered contract data (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CEG contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for CEG contract.

The schedule includes: 1. Fee payment events (FP) if fees are charged 2. Credit event detection (XD) if covered contract defaults 3. Settlement event (STD) after credit event 4. Maturity event (MD) if no credit event occurs

Returns:

EventSchedule with guarantee events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State includes coverage amount calculated from covered contracts.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for CEG contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CEGPayoffFunction instance

Return type:

CEGPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CEG contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CEGStateTransitionFunction instance

Return type:

CEGStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CEG contract with comprehensive event generation.

Generates PRD, FP, XD, STD, and MD events based on covered contract states and credit events observed through the child observer.

Parameters:
Return type:

SimulationHistory

CEC - Credit Enhancement Collateral

Credit Enhancement Collateral (CEC) contract implementation.

This module implements credit enhancement collateral contracts that track collateral value versus covered contract exposure. Similar to margin accounts, CEC contracts compare collateral provided by covering contracts against the exposure from covered contracts.

Key Features:

  • Two sets of contracts: covered and covering

  • Covering contracts provide collateral value

  • Covered contracts represent exposure

  • Compares collateral vs required amount

  • Releases excess or seizes shortfall

  • Credit event triggers evaluation

Example

>>> from jactus.contracts import CreditEnhancementCollateralContract
>>> from jactus.core import ContractAttributes, ActusDateTime
>>> from jactus.observers import ConstantRiskFactorObserver, MockChildContractObserver
>>>
>>> # Create collateral contract for a loan
>>> attrs = ContractAttributes(
...     contract_id="CEC-001",
...     contract_type=ContractType.CEC,
...     contract_role=ContractRole.RPA,
...     status_date=ActusDateTime(2024, 1, 1, 0, 0, 0),
...     maturity_date=ActusDateTime(2029, 1, 1, 0, 0, 0),
...     coverage=1.2,  # 120% collateral requirement
...     credit_enhancement_guarantee_extent="NO",  # Notional only
...     contract_structure='{"CoveredContract": "LOAN-001", "CoveringContract": "STK-001"}',
... )
>>> rf_obs = ConstantRiskFactorObserver(0.03)
>>> child_obs = MockChildContractObserver()
>>> cec = CreditEnhancementCollateralContract(attrs, rf_obs, child_obs)
>>> cashflows = cec.simulate(rf_obs, child_obs)

References

ACTUS Technical Specification v1.1, Section 7.18

class jactus.contracts.cec.CECPayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: BasePayoffFunction

Payoff function for CEC contracts.

CEC payoffs represent collateral settlement (return or seizure).

Parameters:
calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for collateral events.

Parameters:
Returns:

Payoff amount (collateral settlement)

Return type:

Array

class jactus.contracts.cec.CECStateTransitionFunction(day_count_convention=None)[source]

Bases: BaseStateTransitionFunction

State transition function for CEC contracts.

CEC state tracks collateral value vs exposure.

Parameters:

day_count_convention (DayCountConvention | None)

transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Calculate state transition for collateral events.

Parameters:
Returns:

Updated contract state

Return type:

ContractState

class jactus.contracts.cec.CreditEnhancementCollateralContract(*args, **kwargs)[source]

Bases: BaseContract

Credit Enhancement Collateral (CEC) contract.

A collateral contract that tracks collateral value from covering contracts against exposure from covered contracts. When credit events occur or at maturity, compares collateral vs required amount and settles appropriately.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

Any

attributes

Contract terms and conditions

risk_factor_observer

Observer for market rates

child_contract_observer

Observer for covered/covering contracts (required)

__init__(attributes, risk_factor_observer, child_contract_observer=None)[source]

Initialize CEC contract.

Parameters:
Raises:

ValueError – If required attributes are missing or invalid

generate_event_schedule()[source]

Generate event schedule for CEC contract.

The schedule includes: 1. Analysis dates (AD) if specified 2. Credit event detection (XD) if covered contract defaults 3. Settlement event (STD) after credit event or at maturity 4. Maturity event (MD) if no credit event occurs

Returns:

EventSchedule with collateral events

Return type:

EventSchedule

initialize_state()[source]

Initialize contract state at status date.

State includes collateral value compared to required amount.

Returns:

Initial ContractState

Return type:

ContractState

get_payoff_function(event_type)[source]

Get payoff function for CEC contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CECPayoffFunction instance

Return type:

CECPayoffFunction

get_state_transition_function(event_type)[source]

Get state transition function for CEC contract.

Parameters:

event_type (Any) – Type of event (not used, kept for interface compatibility)

Returns:

CECStateTransitionFunction instance

Return type:

CECStateTransitionFunction

simulate(risk_factor_observer=None, child_contract_observer=None, scenario=None, behavior_observers=None)[source]

Simulate CEC contract with comprehensive event generation.

Generates XD, STD, and MD events based on covered/covering contract states and credit events observed through the child observer.

Parameters:
Return type:

SimulationHistory

Contract Functions

Payoff Functions

Payoff function framework for ACTUS contracts.

This module implements the payoff function (POF) framework as defined in Section 2.7 and 3.9 of the ACTUS specification v1.1.

The payoff function f(e, S, M, t, o_rf) calculates the cashflow amount for a contract event, applying contract role sign and FX rate adjustments.

References

ACTUS Technical Specification v1.1: - Section 2.7: Payoff Functions - Section 3.9: Canonical Contract Payoff Function F(x,t) - Section 3.10: Settlement Currency FX Rate X^CURS_CUR(t)

class jactus.functions.payoff.PayoffFunction(*args, **kwargs)[source]

Bases: Protocol

Protocol for payoff functions.

A payoff function calculates the cashflow amount for a contract event. All concrete POF implementations must implement this protocol.

The payoff function signature is:

f(e, S, M, t, o_rf) -> payoff

Where:

e = event type S = pre-event state M = contract attributes t = event time o_rf = risk factor observer

Returns:

Payoff amount as JAX array (scalar)

__call__(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate payoff for an event.

Parameters:
Returns:

Payoff amount as JAX array (scalar)

Return type:

Array

References

ACTUS v1.1 Section 2.7

__init__(*args, **kwargs)
class jactus.functions.payoff.BasePayoffFunction(contract_role, currency, settlement_currency=None)[source]

Bases: ABC

Base class for payoff functions with common logic.

This abstract base class implements the common payoff calculation pipeline: 1. Calculate base payoff amount (contract-specific, abstract) 2. Apply contract role sign R(CNTRL) 3. Apply FX rate X^CURS_CUR(t) if settlement currency differs

Subclasses must implement calculate_payoff() with contract-specific logic.

Parameters:
contract_role

Contract role (RPA, RPL, etc.)

currency

Contract currency

settlement_currency

Settlement currency (None = same as contract currency)

References

ACTUS v1.1 Section 2.7, 3.10

__init__(contract_role, currency, settlement_currency=None)[source]

Initialize base payoff function.

Parameters:

  • contract_role (ContractRole) – Contract role for sign adjustment

  • currency (str) – Contract currency (e.g., “USD”)

  • settlement_currency (str | None) – Settlement currency (None = same as contract)

abstractmethod calculate_payoff(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate base payoff before role sign and FX adjustments.

This is the contract-specific payoff logic that subclasses must implement.

Parameters:
Returns:

Base payoff amount as JAX array (scalar)

Return type:

Array

apply_role_sign(amount)[source]

Apply contract role sign R(CNTRL).

The contract role determines the sign of cashflows: - RPA, LG, BUY, etc.: +1 (receive cashflows) - RPL, ST, SEL, etc.: -1 (pay cashflows)

Parameters:

amount (Array) – Unsigned payoff amount

Returns:

Signed payoff amount

Return type:

Array

Formula:

signed_amount = amount * R(CNTRL)

References

ACTUS v1.1 Table 1 (Contract Role Signs)

apply_fx_rate(amount, time, risk_factor_observer)[source]

Apply FX rate X^CURS_CUR(t) if settlement currency differs.

If the settlement currency differs from the contract currency, the payoff must be converted using the FX rate observed at the event time.

Parameters:
Returns:

Payoff in settlement currency

Return type:

Array

Formula:
If CURS != CUR:

payoff_settlement = payoff_contract * X^CURS_CUR(t)

Else:

payoff_settlement = payoff_contract

References

ACTUS v1.1 Section 3.10

__call__(event_type, state, attributes, time, risk_factor_observer)[source]

Calculate complete payoff with role sign and FX adjustments.

This method implements the complete payoff calculation pipeline: 1. Calculate base payoff (contract-specific) 2. Apply contract role sign 3. Apply FX rate if needed

Parameters:
Returns:

Final payoff amount as JAX array (scalar)

Return type:

Array

References

ACTUS v1.1 Section 2.7

jactus.functions.payoff.settlement_currency_fx_rate(time, contract_currency, settlement_currency, risk_factor_observer)[source]

Get FX rate X^CURS_CUR(t) for settlement currency conversion.

Returns the FX rate to convert from contract currency (CUR) to settlement currency (CURS) at the given time.

Parameters:

  • time (ActusDateTime) – Time at which to observe FX rate

  • contract_currency (str) – Contract currency code (e.g., “USD”)

  • settlement_currency (str | None) – Settlement currency code (None = same as contract)

  • risk_factor_observer (RiskFactorObserver) – Observer for FX rate data

Returns:

FX rate as JAX array (1.0 if currencies are same)

Return type:

Array

Logic:

If settlement_currency is None: return 1.0 If settlement_currency == contract_currency: return 1.0 Otherwise: observe FX rate “contract_currency/settlement_currency”

Example

>>> # Contract in EUR, settled in USD
>>> fx_rate = settlement_currency_fx_rate(
...     time=t,
...     contract_currency="EUR",
...     settlement_currency="USD",
...     risk_factor_observer=observer
... )
>>> # Returns EUR/USD rate, e.g., 1.18

References

ACTUS v1.1 Section 3.10

jactus.functions.payoff.canonical_contract_payoff(contract, time, risk_factor_observer)[source]

Calculate canonical contract payoff F(x, t).

The canonical contract payoff is the sum of all future event payoffs at time t, evaluated using the current risk factor conditions.

This function is used for contract valuation and mark-to-market calculations.

Parameters:
Returns:

Total payoff of all future events as JAX array (scalar)

Return type:

Array

Formula:

F(x, t) = Σ f(e_i, S_i, M, t_i, o_rf) for all events e_i where t_i >= t

Where:

  • e_i = i-th future event

  • S_i = state at event time

  • M = contract attributes

  • t_i = event time

  • o_rf = risk factor observer (frozen at current state)

Note

This uses current risk factor conditions for all future events, which may differ from the actual risk factors at those event times.

Example

>>> contract = PAMContract(attributes)
>>> observer = MockRiskFactorObserver({'LIBOR': {t: 0.03}})
>>> f_xt = canonical_contract_payoff(contract, t, observer)
>>> print(f"Contract value: {f_xt}")

References

ACTUS v1.1 Section 3.9

State Transition Functions

State transition functions for ACTUS contracts.

This module implements the State Transition Function (STF) framework, which defines how contract states evolve through events.

The STF has the signature: STF(e, S_pre, M, t, o_rf) -> S_post where: - e: Event type - S_pre: Pre-event contract state - M: Contract attributes (terms) - t: Event time - o_rf: Risk factor observer

References

ACTUS v1.1 Section 2.8 - State Transition Functions

class jactus.functions.state.StateTransitionFunction(*args, **kwargs)[source]

Bases: Protocol

Protocol for state transition functions.

The state transition function signature is:

STF(e, S_pre, M, t, o_rf) -> S_post

All implementations must be JAX-compatible (pure functions, JIT-compilable).

__call__(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Transition contract state through an event.

Parameters:
Returns:

Post-event contract state

Return type:

ContractState

__init__(*args, **kwargs)
class jactus.functions.state.BaseStateTransitionFunction(day_count_convention=None)[source]

Bases: ABC

Base class for state transition functions with common logic.

This class provides a framework for implementing state transitions with standard patterns like interest accrual, fee accrual, and status date updates.

Subclasses must implement the abstract transition_state() method with contract-specific state transition logic.

Parameters:

day_count_convention (DayCountConvention | None)

__init__(day_count_convention=None)[source]

Initialize state transition function.

Parameters:

day_count_convention (DayCountConvention | None) – Day count convention for time calculations (if None, uses contract’s DCC from attributes)

abstractmethod transition_state(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Implement contract-specific state transition logic.

This method should be implemented by subclasses to define how the contract state transitions for specific event types.

Parameters:
Returns:

Post-event contract state

Return type:

ContractState

Note

This method should typically: 1. Copy the pre-event state 2. Update state variables based on event type 3. Return the new state

Example

>>> def transition_state(self, event_type, state_pre, attributes, time, observer):
...     # Create post-event state by copying pre-event state
...     state_post = state_pre
...
...     # Update state based on event type
...     if event_type == EventType.IP:
...         # Interest payment: reset accrued interest
...         state_post = state_post.replace(ipac=jnp.array(0.0))
...
...     return state_post
update_status_date(state, new_date)[source]

Update contract status date.

Parameters:
Returns:

Updated state with new status date

Return type:

ContractState

accrue_interest(state, attributes, from_date, to_date)[source]

Calculate interest accrued over a period.

Accrual formula: IPAC += NT * IPNR * YearFraction(from, to, DCC)

Parameters:
Returns:

Interest accrued as JAX array

Return type:

jnp.ndarray

References

ACTUS v1.1 Section 3.4 - Interest Accrual

accrue_fees(state, attributes, from_date, to_date, fee_rate, fee_basis)[source]

Calculate fees accrued over a period.

Fee accrual depends on the fee basis: - FeeBasis.A (Absolute): FEAC += FER * YearFraction(from, to, DCC) - FeeBasis.N (Notional): FEAC += NT * FER * YearFraction(from, to, DCC)

Parameters:
Returns:

Fees accrued as JAX array

Return type:

jnp.ndarray

References

ACTUS v1.1 Section 3.5 - Fee Accrual

__call__(event_type, state_pre, attributes, time, risk_factor_observer)[source]

Execute complete state transition pipeline.

This method: 1. Calls the abstract transition_state() method 2. Updates the status date to the event time 3. Returns the post-event state

Parameters:
Returns:

Post-event contract state

Return type:

ContractState

References

ACTUS v1.1 Section 2.8

jactus.functions.state.create_state_pre(tmd, sd, nt, ipnr, ipac=0.0, feac=0.0, nsc=1.0, isc=1.0)[source]

Create a pre-event contract state with default values.

This is a convenience function for creating initial contract states with common default values.

Parameters:

  • tmd (ActusDateTime) – Time of maturity date

  • sd (ActusDateTime) – Status date

  • nt (float) – Notional principal

  • ipnr (float) – Nominal interest rate

  • ipac (float) – Accrued interest (default: 0.0)

  • feac (float) – Accrued fees (default: 0.0)

  • nsc (float) – Notional scaling multiplier (default: 1.0)

  • isc (float) – Interest scaling multiplier (default: 1.0)

Returns:

Initial contract state

Return type:

ContractState

Example

>>> state = create_state_pre(
...     tmd=ActusDateTime(2029, 1, 15, 0, 0, 0),
...     sd=ActusDateTime(2024, 1, 15, 0, 0, 0),
...     nt=100000.0,
...     ipnr=0.05,
... )
jactus.functions.state.validate_state_transition(state_pre, state_post, event_type)[source]

Validate that a state transition is consistent.

Performs basic sanity checks on state transitions: - Status date should not go backwards - Notional should not become negative (for most events) - Scaling factors should remain positive

Parameters:
Returns:

True if transition is valid, False otherwise

Return type:

bool

Example

>>> is_valid = validate_state_transition(state_pre, state_post, EventType.IP)
>>> if not is_valid:
...     raise ValueError("Invalid state transition")

Observers

Risk Factor Observer

Risk factor observer for ACTUS contracts.

This module implements the Risk Factor Observer (O_rf) framework, which provides access to market data and risk factors needed for contract valuation.

The observer has two key methods: - O_rf(i, t, S, M): Observe risk factor i at time t - O_ev(i, k, t, S, M): Observe event-related data

References

ACTUS v1.1 Section 2.9 - Risk Factor Observer

class jactus.observers.risk_factor.RiskFactorObserver(*args, **kwargs)[source]

Bases: Protocol

Protocol for risk factor observers.

The risk factor observer provides access to market data and risk factors needed for contract calculations. It abstracts away the data source (historical data, real-time feeds, simulations, etc.).

All implementations must be JAX-compatible where possible.

observe_risk_factor(identifier, time, state=None, attributes=None)[source]

Observe a risk factor at a specific time.

This is the O_rf(i, t, S, M) function from ACTUS specification.

Parameters:

  • identifier (str) – Risk factor identifier (e.g., “USD/EUR”, “LIBOR-3M”)

  • time (ActusDateTime) – Time at which to observe the risk factor

  • state (ContractState | None) – Current contract state (optional, for state-dependent factors)

  • attributes (ContractAttributes | None) – Contract attributes (optional, for contract-dependent factors)

Returns:

Risk factor value as JAX array

Return type:

jnp.ndarray

Example

>>> observer = DictRiskFactorObserver({"USD/EUR": 1.18})
>>> fx_rate = observer.observe_risk_factor("USD/EUR", time)
>>> # Returns 1.18
observe_event(identifier, event_type, time, state=None, attributes=None)[source]

Observe event-related data.

This is the O_ev(i, k, t, S, M) function from ACTUS specification, where k is the event type.

Parameters:
Returns:

Event-related data (type depends on identifier)

Return type:

Any

Example

>>> observer = DictRiskFactorObserver(...)
>>> rate = observer.observe_event("RESET_RATE", EventType.RR, time)
__init__(*args, **kwargs)
class jactus.observers.risk_factor.BaseRiskFactorObserver(name=None)[source]

Bases: ABC

Base class for risk factor observers with common functionality.

This class provides a framework for implementing risk factor observers with caching, error handling, and JAX compatibility.

Parameters:

name (str | None)

__init__(name=None)[source]

Initialize base risk factor observer.

Parameters:

name (str | None) – Optional name for this observer (for debugging)

observe_risk_factor(identifier, time, state=None, attributes=None)[source]

Observe a risk factor at a specific time.

This method wraps _get_risk_factor with error handling and logging.

Parameters:

  • identifier (str) – Risk factor identifier

  • time (ActusDateTime) – Time at which to observe

  • state (ContractState | None) – Current contract state (optional)

  • attributes (ContractAttributes | None) – Contract attributes (optional)

Returns:

Risk factor value as JAX array

Raises:

KeyError – If risk factor is not found

Return type:

jnp.ndarray

observe_event(identifier, event_type, time, state=None, attributes=None)[source]

Observe event-related data.

This method wraps _get_event_data with error handling and logging.

Parameters:
Returns:

Event-related data

Return type:

Any

class jactus.observers.risk_factor.ConstantRiskFactorObserver(constant_value, name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer that returns constant values.

This is useful for testing or for contracts with fixed risk factors.

Example

>>> observer = ConstantRiskFactorObserver(1.0)
>>> rate = observer.observe_risk_factor("ANY_RATE", time)
>>> # Always returns 1.0
Parameters:

  • constant_value (float)

  • name (str | None)

__init__(constant_value, name=None)[source]

Initialize constant risk factor observer.

Parameters:

  • constant_value (float) – The constant value to return for all risk factors

  • name (str | None) – Optional name for this observer

class jactus.observers.risk_factor.DictRiskFactorObserver(risk_factors, event_data=None, name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer backed by a dictionary.

This is useful for testing or for simple scenarios with a fixed set of risk factors.

Example

>>> data = {
...     "USD/EUR": 1.18,
...     "LIBOR-3M": 0.02,
... }
>>> observer = DictRiskFactorObserver(data)
>>> fx_rate = observer.observe_risk_factor("USD/EUR", time)
>>> # Returns 1.18
Parameters:
__init__(risk_factors, event_data=None, name=None)[source]

Initialize dictionary-backed risk factor observer.

Parameters:

  • risk_factors (dict[str, float]) – Dictionary mapping risk factor identifiers to values

  • event_data (dict[str, Any] | None) – Dictionary mapping event data identifiers to values (optional)

  • name (str | None) – Optional name for this observer

add_risk_factor(identifier, value)[source]

Add or update a risk factor.

Parameters:

  • identifier (str) – Risk factor identifier

  • value (float) – Risk factor value

Return type:

None

add_event_data(identifier, value)[source]

Add or update event data.

Parameters:

  • identifier (str) – Event data identifier

  • value (Any) – Event data value

Return type:

None

class jactus.observers.risk_factor.TimeSeriesRiskFactorObserver(risk_factors, event_data=None, interpolation='step', extrapolation='flat', name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer backed by time series data with interpolation.

Maps identifiers to time-ordered sequences of (ActusDateTime, float) pairs. Supports step (piecewise constant) and linear interpolation, with flat or raising extrapolation behavior.

Example

>>> from jactus.core import ActusDateTime
>>> ts = {
...     "LIBOR-3M": [
...         (ActusDateTime(2024, 1, 1), 0.04),
...         (ActusDateTime(2024, 7, 1), 0.045),
...         (ActusDateTime(2025, 1, 1), 0.05),
...     ]
... }
>>> observer = TimeSeriesRiskFactorObserver(ts)
>>> rate = observer.observe_risk_factor(
...     "LIBOR-3M", ActusDateTime(2024, 4, 1)
... )
>>> # Returns 0.04 (step interpolation: last known value)
Parameters:
__init__(risk_factors, event_data=None, interpolation='step', extrapolation='flat', name=None)[source]

Initialize time series risk factor observer.

Parameters:

  • risk_factors (dict[str, list[tuple[ActusDateTime, float]]]) – Dict mapping identifiers to time-value pairs.

  • event_data (dict[str, list[tuple[ActusDateTime, Any]]] | None) – Optional dict mapping identifiers to time-value pairs for events.

  • interpolation (str) – Interpolation method: “step” (piecewise constant) or “linear”.

  • extrapolation (str) – Extrapolation method: “flat” (nearest endpoint) or “raise” (KeyError).

  • name (str | None) – Optional name for this observer.

class jactus.observers.risk_factor.CurveRiskFactorObserver(curves, reference_date=None, interpolation='linear', name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer for yield/rate curves.

Maps identifiers to tenor-rate curves where each curve is a list of (tenor_years, rate) pairs. Given an observation time, the observer computes the tenor from the reference date and interpolates the curve.

Example

>>> from jactus.core import ActusDateTime
>>> curve = {
...     "USD-YIELD": [
...         (0.25, 0.03),   # 3-month rate
...         (1.0, 0.04),    # 1-year rate
...         (5.0, 0.05),    # 5-year rate
...     ]
... }
>>> observer = CurveRiskFactorObserver(
...     curves=curve,
...     reference_date=ActusDateTime(2024, 1, 1),
... )
>>> rate = observer.observe_risk_factor(
...     "USD-YIELD", ActusDateTime(2024, 7, 1)
... )
Parameters:
__init__(curves, reference_date=None, interpolation='linear', name=None)[source]

Initialize curve risk factor observer.

Parameters:

  • curves (dict[str, list[tuple[float, float]]]) – Dict mapping identifiers to lists of (tenor_years, rate) pairs.

  • reference_date (ActusDateTime | None) – Base date for tenor calculation. Falls back to attributes.status_date if not set.

  • interpolation (str) – Interpolation method: “linear” or “log_linear”.

  • name (str | None) – Optional name for this observer.

class jactus.observers.risk_factor.CallbackRiskFactorObserver(callback, event_callback=None, name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer that delegates to user-provided callables.

Provides maximum flexibility by allowing arbitrary Python functions to produce risk factor values.

Example

>>> import math
>>> def my_rate(identifier: str, time: ActusDateTime) -> float:
...     years = ActusDateTime(2024, 1, 1).years_between(time)
...     return 0.03 + 0.01 * math.log(1 + max(years, 0))
...
>>> observer = CallbackRiskFactorObserver(callback=my_rate)
>>> rate = observer.observe_risk_factor("ANY", ActusDateTime(2025, 1, 1))
Parameters:
__init__(callback, event_callback=None, name=None)[source]

Initialize callback risk factor observer.

Parameters:

  • callback (Callable[[str, ActusDateTime], float]) – Function taking (identifier, time) and returning a float.

  • event_callback (Callable[[str, EventType, ActusDateTime], Any] | None) – Optional function taking (identifier, event_type, time) and returning event data.

  • name (str | None) – Optional name for this observer.

class jactus.observers.risk_factor.CompositeRiskFactorObserver(observers, name=None)[source]

Bases: BaseRiskFactorObserver

Risk factor observer that chains multiple observers with fallback.

Tries each observer in order and returns the first successful result. If an observer raises KeyError, the next one is tried. Other exceptions propagate immediately.

Example

>>> ts_observer = TimeSeriesRiskFactorObserver({"LIBOR-3M": [...]})
>>> fallback = ConstantRiskFactorObserver(0.0)
>>> composite = CompositeRiskFactorObserver([ts_observer, fallback])
>>> # Uses ts_observer for "LIBOR-3M", falls back to constant for others
Parameters:
__init__(observers, name=None)[source]

Initialize composite risk factor observer.

Parameters:

  • observers (list[RiskFactorObserver]) – List of observers to try in order. Must not be empty.

  • name (str | None) – Optional name for this observer.

Raises:

ValueError – If observers list is empty.

class jactus.observers.risk_factor.JaxRiskFactorObserver(risk_factors, default_value=0.0)[source]

Bases: object

Fully JAX-compatible risk factor observer.

This observer is designed for use with jax.jit and jax.grad. It uses integer indices instead of string identifiers and stores all data in JAX arrays.

Key features: - Pure functions (no side effects) - JIT-compilable - Differentiable with jax.grad - Vectorized with jax.vmap - No Python control flow in hot paths

Example

>>> # Create observer with 3 risk factors
>>> risk_factors = jnp.array([1.18, 0.05, 100000.0])
>>> observer = JaxRiskFactorObserver(risk_factors)
>>>
>>> # Observe risk factor at index 0 (e.g., FX rate)
>>> fx_rate = observer.get(0)  # Returns 1.18
>>>
>>> # Use with jax.grad for sensitivities
>>> def contract_value(risk_factors):
...     obs = JaxRiskFactorObserver(risk_factors)
...     fx = obs.get(0)
...     rate = obs.get(1)
...     notional = obs.get(2)
...     return notional * rate * fx
>>>
>>> # Compute gradient (sensitivities)
>>> sensitivities = jax.grad(contract_value)(risk_factors)
>>> # sensitivities[0] = d(value)/d(fx_rate)
>>> # sensitivities[1] = d(value)/d(rate)
>>> # sensitivities[2] = d(value)/d(notional)

Note

This observer does not implement the RiskFactorObserver protocol because the protocol uses string identifiers which are not JAX-compatible. Instead, it provides a simpler API with integer indices.

References

ACTUS v1.1 Section 2.9 - Risk Factor Observer

Parameters:

  • risk_factors (jnp.ndarray)

  • default_value (float | jnp.ndarray)

__init__(risk_factors, default_value=0.0)[source]

Initialize JAX-compatible risk factor observer.

Parameters:

  • risk_factors (Array) – JAX array of risk factor values, indexed by integer

  • default_value (float | Array) – Default value to return for out-of-bounds indices

Example

>>> # Create observer with FX rate, interest rate, notional
>>> risk_factors = jnp.array([1.18, 0.05, 100000.0])
>>> observer = JaxRiskFactorObserver(risk_factors)
get(index)[source]

Get risk factor value at the given index.

This method is JIT-compilable and differentiable.

Parameters:

index (int) – Integer index of the risk factor

Returns:

Risk factor value as JAX array

Return type:

Array

Example

>>> observer = JaxRiskFactorObserver(jnp.array([1.18, 0.05]))
>>> fx_rate = observer.get(0)  # Returns 1.18
>>> rate = observer.get(1)     # Returns 0.05

Note

Uses safe indexing with bounds checking via JAX operations. Out-of-bounds indices return the default value.

get_batch(indices)[source]

Get multiple risk factors at once (vectorized).

This is useful for batch operations and is vmappable.

Parameters:

indices (Array) – Array of integer indices

Returns:

Array of risk factor values

Return type:

Array

Example

>>> observer = JaxRiskFactorObserver(jnp.array([1.18, 0.05, 100000.0]))
>>> indices = jnp.array([0, 2])  # Get FX rate and notional
>>> values = observer.get_batch(indices)
>>> # Returns jnp.array([1.18, 100000.0])
update(index, value)[source]

Create new observer with updated risk factor value.

This is a pure function - it returns a new observer without modifying the original.

Parameters:

  • index (int) – Index of risk factor to update

  • value (float) – New value

Returns:

New JaxRiskFactorObserver with updated value

Return type:

JaxRiskFactorObserver

Example

>>> observer = JaxRiskFactorObserver(jnp.array([1.18, 0.05]))
>>> new_observer = observer.update(0, 1.20)  # Update FX rate
>>> new_observer.get(0)  # Returns 1.20
>>> observer.get(0)      # Still returns 1.18 (original unchanged)
update_batch(indices, values)[source]

Create new observer with multiple updated risk factors.

Parameters:

  • indices (Array) – Array of indices to update

  • values (Array) – Array of new values

Returns:

New JaxRiskFactorObserver with updated values

Return type:

JaxRiskFactorObserver

Example

>>> observer = JaxRiskFactorObserver(jnp.array([1.18, 0.05, 100000.0]))
>>> new_observer = observer.update_batch(
...     jnp.array([0, 1]),
...     jnp.array([1.20, 0.06])
... )
to_array()[source]

Get all risk factors as a JAX array.

Returns:

Array of all risk factor values

Return type:

Array

Example

>>> observer = JaxRiskFactorObserver(jnp.array([1.18, 0.05]))
>>> observer.to_array()  # Returns jnp.array([1.18, 0.05])
static from_dict(mapping, size=None, default_value=0.0)[source]

Create observer from a dictionary mapping indices to values.

This is a convenience method for initialization. The observer itself remains fully JAX-compatible.

Parameters:

  • mapping (dict[int, float]) – Dictionary mapping integer indices to float values

  • size (int | None) – Size of risk factor array (if None, uses max index + 1)

  • default_value (float) – Default value for unspecified indices

Returns:

New JaxRiskFactorObserver

Return type:

JaxRiskFactorObserver

Example

>>> observer = JaxRiskFactorObserver.from_dict({
...     0: 1.18,    # FX rate
...     1: 0.05,    # Interest rate
...     2: 100000.0 # Notional
... })

Child Contract Observer

Child contract observer for composite ACTUS contracts.

This module implements observers for child contracts in composite contract structures. In ACTUS, composite contracts can observe events, states, and attributes of their child contracts to determine their own behavior.

References

ACTUS v1.1 Section 2.10 - Child Contract Observer ACTUS v1.1 Section 3.4 - Composite Contract Types

class jactus.observers.child_contract.ChildContractObserver(*args, **kwargs)[source]

Bases: Protocol

Protocol for observing child contract data in composite contracts.

The child contract observer provides access to events, states, and attributes of child contracts within a composite structure. This enables parent contracts to make decisions based on child contract behavior.

Methods correspond to ACTUS observer functions: - observe_events: U_ev(i, t, a) - Observe child events at time t - observe_state: U_sv(i, t, x, a) - Observe child state at time t - observe_attribute: U_ca(i, x) - Observe child attribute value

Example

>>> observer = MockChildContractObserver()
>>> observer.register_child("loan1", child_contract)
>>> events = observer.observe_events("loan1", current_time, attributes)
>>> state = observer.observe_state("loan1", current_time, None, attributes)
>>> notional = observer.observe_attribute("loan1", "notional_principal")

Note

This protocol uses runtime_checkable to allow isinstance() checks. Implementations should handle missing child contracts gracefully.

References

ACTUS v1.1 Section 2.10 - Child Contract Observer

observe_events(identifier, time, attributes=None)[source]

Observe events from a child contract.

Returns all events from the identified child contract that occur at or after the specified time. This allows parent contracts to react to child contract events.

Parameters:

  • identifier (str) – Child contract identifier

  • time (ActusDateTime) – Observation time

  • attributes (ContractAttributes | None) – Optional parent contract attributes for filtering

Returns:

List of child contract events

Return type:

list[ContractEvent]

Example

>>> events = observer.observe_events("child1", t0, parent_attrs)
>>> principal_events = [e for e in events if e.event_type == EventType.PR]
observe_state(identifier, time, state=None, attributes=None)[source]

Observe state from a child contract.

Returns the state of the identified child contract at the specified time. This allows parent contracts to access child contract state variables.

Parameters:

  • identifier (str) – Child contract identifier

  • time (ActusDateTime) – Observation time

  • state (ContractState | None) – Optional parent state for context

  • attributes (ContractAttributes | None) – Optional parent attributes for context

Returns:

Child contract state at the specified time

Return type:

ContractState

Example

>>> child_state = observer.observe_state("child1", t0, parent_state, parent_attrs)
>>> child_notional = child_state.nt
observe_attribute(identifier, attribute_name)[source]

Observe an attribute value from a child contract.

Returns the value of a specific attribute from the identified child contract. This allows parent contracts to access child contract configuration.

Parameters:

  • identifier (str) – Child contract identifier

  • attribute_name (str) – Name of the attribute to observe

Returns:

Attribute value as JAX array

Return type:

Array

Example

>>> notional = observer.observe_attribute("child1", "notional_principal")
>>> rate = observer.observe_attribute("child1", "nominal_interest_rate")
__init__(*args, **kwargs)
class jactus.observers.child_contract.BaseChildContractObserver[source]

Bases: ABC

Abstract base class for child contract observers.

Provides a common implementation pattern with error handling and validation. Subclasses implement the abstract methods to define observation behavior.

Example

>>> class MyObserver(BaseChildContractObserver):
...     def _get_events(self, identifier, time, attributes):
...         return self.children[identifier].get_events()
...     def _get_state(self, identifier, time, state, attributes):
...         return self.children[identifier].get_state(time)
...     def _get_attribute(self, identifier, attribute_name):
...         return getattr(self.children[identifier].attributes, attribute_name)

References

ACTUS v1.1 Section 2.10 - Child Contract Observer

observe_events(identifier, time, attributes=None)[source]

Observe events from a child contract with error handling.

Wrapper that adds validation and error handling around _get_events.

Parameters:
Returns:

List of child contract events

Raises:
Return type:

list[ContractEvent]

observe_state(identifier, time, state=None, attributes=None)[source]

Observe state from a child contract with error handling.

Wrapper that adds validation and error handling around _get_state.

Parameters:
Returns:

Child contract state

Raises:
Return type:

ContractState

observe_attribute(identifier, attribute_name)[source]

Observe attribute from a child contract with error handling.

Wrapper that adds validation and error handling around _get_attribute.

Parameters:

  • identifier (str) – Child contract identifier

  • attribute_name (str) – Attribute name

Returns:

Attribute value as JAX array

Raises:
Return type:

Array

class jactus.observers.child_contract.MockChildContractObserver[source]

Bases: BaseChildContractObserver

Mock implementation for child contract observation.

This observer stores child contract data in dictionaries and provides simple observation capabilities. Useful for testing and development.

child_events

Dictionary mapping child IDs to event lists

child_states

Dictionary mapping child IDs to states

child_attributes

Dictionary mapping child IDs to attribute dicts

Example

>>> observer = MockChildContractObserver()
>>> # Register child contract data
>>> observer.register_child("loan1",
...     events=[event1, event2],
...     state=loan_state,
...     attributes={"notional_principal": 100000.0}
... )
>>> # Observe child data
>>> events = observer.observe_events("loan1", t0)
>>> state = observer.observe_state("loan1", t0)
>>> notional = observer.observe_attribute("loan1", "notional_principal")

Note

This is a simple mock implementation. Real implementations would integrate with actual contract simulation engines.

References

ACTUS v1.1 Section 2.10 - Child Contract Observer

__init__()[source]

Initialize empty child contract observer.

Return type:

None

register_child(identifier, events=None, state=None, attributes=None)[source]

Register a child contract with its data.

Parameters:

  • identifier (str) – Child contract identifier

  • events (list[ContractEvent] | None) – Optional list of child events

  • state (ContractState | None) – Optional child state

  • attributes (dict[str, float] | None) – Optional dictionary of attribute name -> value

Return type:

None

Example

>>> observer.register_child("child1",
...     events=[ContractEvent(...)],
...     state=ContractState(...),
...     attributes={"notional_principal": 100000.0}
... )
apply_conditions(attributes, overrides)[source]

Apply conditional attribute overrides to contract attributes.

This method temporarily modifies contract attributes based on child contract observations. Used in composite contracts where parent attributes depend on child state.

Parameters:
Returns:

New ContractAttributes with overrides applied

Return type:

ContractAttributes

Example

>>> # Override notional based on child contract
>>> child_notional = observer.observe_attribute("child1", "notional_principal")
>>> new_attrs = observer.apply_conditions(
...     parent_attrs,
...     {"notional_principal": float(child_notional)}
... )

Note

This creates a new ContractAttributes instance rather than modifying the original (immutability).

class jactus.observers.child_contract.SimulatedChildContractObserver[source]

Bases: BaseChildContractObserver

Child contract observer backed by full simulation histories.

Stores the complete event history from simulated child contracts and provides time-aware state lookups (returns state at the most recent event at or before the query time).

Example

>>> from jactus.contracts import create_contract
>>> observer = SimulatedChildContractObserver()
>>> child_result = child_contract.simulate()
>>> observer.register_simulation("child1", child_result.events, child_attrs)
>>> state = observer.observe_state("child1", query_time)
__init__()[source]

Initialize empty observer.

Return type:

None

register_simulation(identifier, events, attributes=None, initial_state=None)[source]

Register a child contract’s simulation results.

Parameters:

  • identifier (str) – Child contract identifier

  • events (list[ContractEvent]) – Full list of simulated events (with state_pre/state_post)

  • attributes (ContractAttributes | None) – Optional child contract attributes

  • initial_state (ContractState | None) – Optional initial state (for queries before first event)

Return type:

None

Behavioral Observer Framework

Behavioral risk factor observers for ACTUS contracts.

This module implements the Behavioral Risk Factor Observer framework, which provides state-dependent risk modeling capabilities. Unlike market risk factor observers that return values based solely on identifiers and time, behavioral observers are aware of the contract’s internal state (notional, interest rate, age, performance status, etc.) and can dynamically inject events into the simulation timeline.

Key concepts:

  • CalloutEvent: An event that a behavioral model requests be added to the simulation schedule. When the simulation reaches that time, it calls back to the behavioral observer with the current contract state.

  • BehaviorRiskFactorObserver: Protocol extending RiskFactorObserver with a contract_start method that returns callout events.

  • BaseBehaviorRiskFactorObserver: Abstract base class providing the framework for implementing concrete behavioral models.

This architecture mirrors the ACTUS risk service’s separation of market risk (pure time-series observation) from behavioral risk (state-dependent models like prepayment and deposit transaction behavior).

References

ACTUS Risk Service v2.0 - BehaviorRiskModelProvider interface ACTUS Technical Specification v1.1, Section 2.9 - Risk Factor Observer

class jactus.observers.behavioral.CalloutEvent(model_id, time, callout_type, metadata=None)[source]

Bases: object

An event requested by a behavioral risk model for injection into the simulation schedule.

Behavioral models return these from contract_start() to register observation times. When the simulation engine reaches the specified time, it evaluates the behavioral observer with the current contract state.

Parameters:
model_id

Identifier of the behavioral model requesting this event (e.g., "ppm01" for a prepayment model).

Type:

str

time

The time at which this observation event should occur.

Type:

ActusDateTime

callout_type

Type code indicating the nature of the callout. Common types include: - "MRD" (Multiplicative Reduction Delta) for prepayment models - "AFD" (Absolute Funded Delta) for deposit transaction models

Type:

str

metadata

Optional additional data the model needs at callout time (e.g., reference rate identifier, surface parameters).

Type:

dict[str, Any] | None

Example

>>> from jactus.core import ActusDateTime
>>> event = CalloutEvent(
...     model_id="prepayment-model-01",
...     time=ActusDateTime(2025, 6, 1),
...     callout_type="MRD",
... )
model_id: str
time: ActusDateTime
callout_type: str
metadata: dict[str, Any] | None = None
__init__(model_id, time, callout_type, metadata=None)
Parameters:
Return type:

None

class jactus.observers.behavioral.BehaviorRiskFactorObserver(*args, **kwargs)[source]

Bases: Protocol

Protocol for behavioral risk factor observers.

Extends the standard risk factor observer concept with state-dependent observation and dynamic event injection. Behavioral observers:

  1. Are called at the start of contract simulation via contract_start() to register observation times (callout events).

  2. At those times, are called via observe_risk_factor() with the full contract state, enabling state-dependent calculations.

This mirrors the ACTUS risk service’s BehaviorRiskModelProvider interface, which separates behavioral models from pure market data.

All implementations should be JAX-compatible where possible.

observe_risk_factor(identifier, time, state=None, attributes=None)[source]

Observe a behavioral risk factor at a specific time.

Unlike market risk observers, behavioral observers typically use the state and attributes parameters to compute their output. For example, a prepayment model computes the spread between the contract’s nominal interest rate and the current market rate.

Parameters:

  • identifier (str) – Risk factor identifier (e.g., model ID).

  • time (ActusDateTime) – Time at which to observe.

  • state (ContractState | None) – Current contract state (used for state-dependent factors).

  • attributes (ContractAttributes | None) – Contract attributes (used for contract-dependent factors).

Returns:

Risk factor value as JAX array.

Return type:

jnp.ndarray

observe_event(identifier, event_type, time, state=None, attributes=None)[source]

Observe event-related behavioral data.

Parameters:
Returns:

Event-related data.

Return type:

Any

contract_start(attributes)[source]

Called at the start of contract simulation.

The behavioral model inspects the contract attributes to determine when it needs to be evaluated during the simulation. It returns a list of CalloutEvent objects that the simulation engine merges into the event schedule.

For example, a prepayment model might return semi-annual observation events over the life of the contract.

Parameters:

attributes (ContractAttributes) – Contract attributes (terms and conditions).

Returns:

List of callout events to inject into the simulation schedule. May be empty if the model does not need scheduled observations.

Return type:

list[CalloutEvent]

Example

>>> events = observer.contract_start(attributes)
>>> for e in events:
...     print(f"{e.model_id} @ {e.time}: {e.callout_type}")
__init__(*args, **kwargs)
class jactus.observers.behavioral.BaseBehaviorRiskFactorObserver(name=None)[source]

Bases: BaseRiskFactorObserver

Abstract base class for behavioral risk factor observers.

Provides the framework for implementing state-dependent risk models that can inject callout events into the simulation schedule. Subclasses must implement:

  • _get_risk_factor(): State-aware risk factor computation

  • _get_event_data(): Event data retrieval

  • contract_start(): Callout event generation

Example

>>> class MyBehavioralModel(BaseBehaviorRiskFactorObserver):
...     def _get_risk_factor(self, identifier, time, state, attributes):
...         # Use state.ipnr (contract rate) in computation
...         spread = float(state.ipnr) - 0.03
...         return jnp.array(max(spread, 0.0) * 0.1)
...
...     def _get_event_data(self, identifier, event_type, time, state, attributes):
...         raise KeyError("No event data")
...
...     def contract_start(self, attributes):
...         return [CalloutEvent("my-model", attributes.status_date, "MRD")]
Parameters:

name (str | None)

__init__(name=None)[source]

Initialize behavioral risk factor observer.

Parameters:

name (str | None) – Optional name for this observer (for debugging).

abstractmethod contract_start(attributes)[source]

Called at the start of contract simulation.

Must be implemented by subclasses to return callout events that should be injected into the simulation schedule.

Parameters:

attributes (ContractAttributes) – Contract attributes.

Returns:

List of CalloutEvent objects.

Return type:

list[CalloutEvent]

Prepayment Surface Observer

Prepayment behavioral risk model for ACTUS contracts.

This module implements a 2D surface-based prepayment model that computes prepayment rates as a function of:

  • Spread (dimension 1): The difference between the contract’s nominal interest rate and the current market reference rate. A positive spread means the borrower has an incentive to refinance.

  • Loan age (dimension 2): Time elapsed since the initial exchange date. Prepayment behavior typically follows a seasoning pattern, peaking in the middle years of a loan’s life.

The model returns a Multiplicative Reduction Delta (MRD) — a fraction by which the notional principal is reduced at each prepayment observation time.

This mirrors the TwoDimensionalPrepaymentModel from the ACTUS risk service, with the addition of JAX compatibility for automatic differentiation.

References

ACTUS Risk Service v2.0 - TwoDimensionalPrepaymentModel ACTUS Technical Specification v1.1 - PP (Principal Prepayment) events

class jactus.observers.prepayment.PrepaymentSurfaceObserver(surface, market_rate_id=None, market_observer=None, fixed_market_rate=0.0, prepayment_cycle='6M', model_id='prepayment-model', name=None)[source]

Bases: BaseBehaviorRiskFactorObserver

Prepayment model using a 2D surface (spread x loan age).

At each prepayment observation time, the model: 1. Computes the spread = state.ipnr - market_rate(time) 2. Computes the loan age = years since attributes.initial_exchange_date 3. Looks up the prepayment rate from the 2D surface 4. Returns the prepayment rate as a JAX array

The market reference rate is obtained from a companion market observer or from a fixed reference rate.

Parameters:

  • surface (Surface2D)

  • market_rate_id (str | None)

  • market_observer (Any | None)

  • fixed_market_rate (float)

  • prepayment_cycle (str)

  • model_id (str)

  • name (str | None)

surface

2D surface mapping (spread, age) to prepayment rate.

market_rate_id

Identifier for the market reference rate (e.g., "UST-5Y"). If not provided, fixed_market_rate is used.

market_observer

Optional companion market risk factor observer for fetching the current market rate.

fixed_market_rate

Fixed market rate used when no market observer is provided (default 0.0).

prepayment_cycle

Cycle string for prepayment event frequency (e.g., "6M" for semi-annual). Used by contract_start() to generate callout events.

model_id

Identifier for this prepayment model instance.

Example

>>> import jax.numpy as jnp
>>> from jactus.utilities.surface import Surface2D
>>> surface = Surface2D(
...     x_margins=jnp.array([-5.0, 0.0, 1.0, 2.0, 3.0]),
...     y_margins=jnp.array([0.0, 1.0, 2.0, 3.0, 5.0, 10.0]),
...     values=jnp.array([
...         [0.00, 0.00, 0.00, 0.00, 0.00, 0.00],  # spread=-5%
...         [0.00, 0.00, 0.01, 0.00, 0.00, 0.00],  # spread= 0%
...         [0.00, 0.01, 0.02, 0.00, 0.00, 0.00],  # spread= 1%
...         [0.00, 0.02, 0.05, 0.03, 0.005, 0.00], # spread= 2%
...         [0.01, 0.05, 0.10, 0.07, 0.02, 0.00],  # spread= 3%
...     ]),
... )
>>> observer = PrepaymentSurfaceObserver(
...     surface=surface,
...     fixed_market_rate=0.04,
...     prepayment_cycle="6M",
... )
__init__(surface, market_rate_id=None, market_observer=None, fixed_market_rate=0.0, prepayment_cycle='6M', model_id='prepayment-model', name=None)[source]

Initialize prepayment surface observer.

Parameters:

  • surface (Surface2D) – 2D surface mapping (spread, age) to prepayment rate.

  • market_rate_id (str | None) – Identifier for the market reference rate.

  • market_observer (Any | None) – Optional market risk factor observer.

  • fixed_market_rate (float) – Fixed market rate when no observer is provided.

  • prepayment_cycle (str) – Cycle for prepayment observation frequency.

  • model_id (str) – Unique model identifier.

  • name (str | None) – Optional observer name for debugging.

contract_start(attributes)[source]

Generate prepayment observation events over the contract life.

Creates callout events at the specified prepayment cycle interval from the initial exchange date to the maturity date.

Parameters:

attributes (ContractAttributes) – Contract attributes.

Returns:

List of CalloutEvent objects with callout_type "MRD".

Return type:

list[CalloutEvent]

Deposit Transaction Observer

Deposit transaction behavioral risk model for ACTUS contracts.

This module implements a deposit transaction model for UMP (Undefined Maturity Profile) contracts. It models deposit inflows and outflows as a function of:

  • Dimension 1: Contract identifier (which specific deposit account)

  • Dimension 2: Date/time of the transaction

The model uses a labeled 2D surface where the x-axis is the contract ID and the y-axis is a date label, returning the transaction amount (an Absolute Funded Delta — the absolute change in the deposit balance).

This mirrors the TwoDimensionalDepositTrxModel from the ACTUS risk service.

References

ACTUS Risk Service v2.0 - TwoDimensionalDepositTrxModel ACTUS Technical Specification v1.1 - UMP contract type

class jactus.observers.deposit_transaction.DepositTransactionObserver(transactions, model_id='deposit-trx-model', name=None)[source]

Bases: BaseBehaviorRiskFactorObserver

Deposit transaction model for UMP contracts.

Models deposit inflows and outflows using a schedule of known or projected transactions. Each contract identifier has its own transaction schedule, looked up from a labeled surface or a simpler time-series mapping.

At each transaction observation time, the model returns the Absolute Funded Delta (AFD) — the change in the deposit balance (positive for inflows, negative for outflows).

Parameters:
transactions

Dictionary mapping contract identifiers to a sorted list of (ActusDateTime, float) pairs (time, amount).

model_id

Identifier for this deposit model instance.

Example

>>> from jactus.core import ActusDateTime
>>> observer = DepositTransactionObserver(
...     transactions={
...         "DEPOSIT-001": [
...             (ActusDateTime(2024, 1, 15), 10000.0),
...             (ActusDateTime(2024, 4, 15), -2000.0),
...             (ActusDateTime(2024, 7, 15), 5000.0),
...         ],
...         "DEPOSIT-002": [
...             (ActusDateTime(2024, 2, 1), 50000.0),
...             (ActusDateTime(2024, 8, 1), -10000.0),
...         ],
...     },
... )
__init__(transactions, model_id='deposit-trx-model', name=None)[source]

Initialize deposit transaction observer.

Parameters:

  • transactions (dict[str, list[tuple[ActusDateTime, float]]]) – Mapping of contract IDs to transaction schedules. Each schedule is a list of (time, amount) pairs, sorted by time.

  • model_id (str) – Unique model identifier.

  • name (str | None) – Optional observer name for debugging.

classmethod from_labeled_surface(surface, date_parser=None, model_id='deposit-trx-model', name=None)[source]

Create from a LabeledSurface2D.

The x-axis labels are contract IDs and y-axis labels are date strings.

Parameters:

  • surface (LabeledSurface2D) – Labeled 2D surface with contract IDs and date labels.

  • date_parser (Any) – Optional callable to parse date labels into ActusDateTime. Defaults to ActusDateTime.from_iso.

  • model_id (str) – Unique model identifier.

  • name (str | None) – Optional observer name.

Returns:

New DepositTransactionObserver instance.

Return type:

DepositTransactionObserver

contract_start(attributes)[source]

Generate callout events for all scheduled transactions.

Returns a callout event for each transaction time associated with the contract’s contract_id.

Parameters:

attributes (ContractAttributes) – Contract attributes (uses contract_id to look up the transaction schedule).

Returns:

List of CalloutEvent objects with callout_type "AFD".

Return type:

list[CalloutEvent]

get_transaction_schedule(contract_id)[source]

Get the full transaction schedule for a contract.

Parameters:

contract_id (str) – Contract identifier.

Returns:

Sorted list of (time, amount) pairs.

Raises:

KeyError – If contract identifier not found.

Return type:

list[tuple[ActusDateTime, float]]

Scenario Management

Scenario management for ACTUS contract simulation.

A Scenario bundles together all the risk factor observers (both market and behavioral) needed for a simulation run into a single, named, reusable configuration.

This provides a higher-level abstraction over individual observers, enabling: - Named, reusable simulation configurations - Consistent bundling of market data with behavioral models - Easy scenario comparison (base case vs. stress scenarios)

The scenario automatically composes its market and behavioral observers using a CompositeRiskFactorObserver so that a single unified observer can be passed to the simulation engine.

References

ACTUS Risk Service v2.0 - Scenario API

class jactus.observers.scenario.Scenario(scenario_id, description='', market_observers=<factory>, behavior_observers=<factory>)[source]

Bases: object

A named simulation scenario bundling market and behavioral observers.

A scenario acts as a simulation environment, declaring which market data sources and behavioral models are available. It provides a unified risk factor observer that the simulation engine can use directly.

Parameters:
scenario_id

Unique identifier for this scenario.

Type:

str

description

Human-readable description.

Type:

str

market_observers

Dictionary mapping identifiers to market risk factor observers. These handle pure market data lookups (time series, curves, constants).

Type:

dict[str, jactus.observers.risk_factor.RiskFactorObserver]

behavior_observers

Dictionary mapping identifiers to behavioral risk factor observers. These are state-aware and can inject callout events into the simulation timeline.

Type:

dict[str, jactus.observers.behavioral.BehaviorRiskFactorObserver]

Example

>>> from jactus.observers import (
...     ConstantRiskFactorObserver,
...     TimeSeriesRiskFactorObserver,
... )
>>> from jactus.observers.prepayment import PrepaymentSurfaceObserver
>>>
>>> scenario = Scenario(
...     scenario_id="base-case",
...     description="Base case with 5Y UST falling and moderate prepayment",
...     market_observers={
...         "rates": TimeSeriesRiskFactorObserver({
...             "UST-5Y": [
...                 (ActusDateTime(2024, 1, 1), 0.045),
...                 (ActusDateTime(2025, 1, 1), 0.035),
...             ],
...         }),
...     },
...     behavior_observers={
...         "prepayment": prepayment_observer,
...     },
... )
>>> # Get unified observer for simulation
>>> observer = scenario.get_observer()
>>> contract.simulate(risk_factor_observer=observer)
scenario_id: str
description: str = ''
market_observers: dict[str, RiskFactorObserver]
behavior_observers: dict[str, BehaviorRiskFactorObserver]
get_observer()[source]

Get a unified risk factor observer that combines all market observers.

Returns a CompositeRiskFactorObserver that chains all market observers in order, providing a single observer for the simulation engine.

If only one market observer is configured, it is returned directly. If no market observers are configured, raises ValueError.

Returns:

Unified RiskFactorObserver.

Raises:

ValueError – If no market observers are configured.

Return type:

RiskFactorObserver

get_callout_events(attributes)[source]

Collect callout events from all behavioral observers.

Calls contract_start() on each behavioral observer and aggregates the returned callout events.

Parameters:

attributes (ContractAttributes) – Contract attributes.

Returns:

Combined list of callout events from all behavioral models, sorted by time.

Return type:

list[CalloutEvent]

add_market_observer(identifier, observer)[source]

Add or replace a market risk factor observer.

Parameters:

  • identifier (str) – Observer identifier.

  • observer (RiskFactorObserver) – Market risk factor observer.

Return type:

None

add_behavior_observer(identifier, observer)[source]

Add or replace a behavioral risk factor observer.

Parameters:
Return type:

None

list_risk_factors()[source]

List all configured risk factor sources.

Returns:

Dictionary mapping identifiers to their observer type names.

Return type:

dict[str, str]

__init__(scenario_id, description='', market_observers=<factory>, behavior_observers=<factory>)
Parameters:
Return type:

None

Portfolio API

Unified portfolio simulation API for mixed contract types.

Accepts a portfolio of contracts with mixed types, groups them by type, dispatches to each type’s batch kernel, and returns per-contract results.

Contracts without a dedicated array-mode implementation (CLM, UMP, SWAPS, CAPFL, CEG, CEC) fall back to the scalar Python simulation path.

Example:

from jactus.contracts.portfolio import simulate_portfolio

contracts = [
    (pam_attrs, rf_obs),
    (lam_attrs, rf_obs),
    (csh_attrs, rf_obs),
    (optns_attrs, rf_obs),
]
result = simulate_portfolio(contracts)
# result["total_cashflows"]  -> jnp.array of shape (4,)
# result["payoffs"]          -> list of per-contract payoff arrays
jactus.contracts.portfolio.simulate_portfolio(contracts, discount_rate=None)[source]

Simulate a mixed-type portfolio using optimal batch strategies.

Groups contracts by type, dispatches to each type’s batch kernel, and falls back to the scalar Python path for unsupported types.

Parameters:

  • contracts (list[tuple[ContractAttributes, RiskFactorObserver]]) – List of (attributes, rf_observer) pairs. Contract types may be mixed.

  • discount_rate (float | None) – If provided, compute present values (passed to each type’s portfolio function where supported).

Returns:

  • total_cashflows: jnp.ndarray of shape (N,) — total cashflow for each contract in input order.

  • num_contracts: Total number of contracts.

  • batch_contracts: Number of contracts simulated via batch kernels.

  • fallback_contracts: Number of contracts simulated via the scalar Python path.

  • types_used: Set of ContractType values present.

  • per_type_results: Dict mapping ContractType to the raw result dict from each type’s portfolio function (only for batch-simulated types).

Return type:

Dict with

Simulation Engine

CLI

JACTUS command-line interface.

Built with Typer. Entry point: jactus = "jactus.cli:app"

class jactus.cli.CLIState(output=OutputFormat.TEXT, pretty=True, no_color=False, log_level='WARNING')[source]

Bases: object

Mutable state shared across all CLI commands via typer.Context.obj.

Parameters:

  • output (OutputFormat)

  • pretty (bool)

  • no_color (bool)

  • log_level (str)

output: OutputFormat = 'text'
pretty: bool = True
no_color: bool = False
log_level: str = 'WARNING'
__init__(output=OutputFormat.TEXT, pretty=True, no_color=False, log_level='WARNING')
Parameters:

  • output (OutputFormat)

  • pretty (bool)

  • no_color (bool)

  • log_level (str)

Return type:

None

jactus.cli.get_state()[source]

Return the global CLI state.

Return type:

CLIState

jactus.cli.load_attrs(attrs, stdin)[source]

Load contract attributes from inline JSON, a file path, or stdin.

Parameters:
Return type:

dict[str, Any]

jactus.cli.parse_datetime(value)[source]

Parse an ISO date string to ActusDateTime.

Parameters:

value (str)

Return type:

ActusDateTime

jactus.cli.prepare_attributes(attributes)[source]

Convert string values in attributes to proper JACTUS types.

Parameters:

attributes (dict[str, Any])

Return type:

dict[str, Any]

jactus.cli.main(ctx, output=<typer.models.OptionInfo object>, pretty=<typer.models.OptionInfo object>, no_color=<typer.models.OptionInfo object>, log_level=<typer.models.OptionInfo object>, version=<typer.models.OptionInfo object>)[source]

JACTUS — JAX ACTUS Financial Contract Simulation.

Parameters:

  • ctx (Context)

  • output (str | None)

  • pretty (bool)

  • no_color (bool)

  • log_level (str)

  • version (bool | None)

Return type:

None

Exceptions

Custom exception classes for ACTUS-specific errors.

This module defines a hierarchy of exceptions used throughout the actus_jax package. All exceptions inherit from ActusException, which provides common functionality for error handling and context preservation.

exception jactus.exceptions.ActusException(message, context=None)[source]

Bases: Exception

Base exception for all ACTUS-related errors.

This is the base class for all custom exceptions in the actus_jax package. It provides common functionality for storing context information about errors.

Parameters:
Return type:

None

message

Human-readable error description

context

Additional context information about the error

__init__(message, context=None)[source]

Initialize the exception.

Parameters:

  • message (str) – Human-readable error description

  • context (dict[str, Any] | None) – Optional dictionary with additional error context (e.g., contract_id, event_type, timestamp)

Return type:

None

__str__()[source]

Return string representation of the exception.

Return type:

str

exception jactus.exceptions.ContractValidationError(message, context=None)[source]

Bases: ActusException

Exception raised for invalid contract attributes or configuration.

This exception should be raised when: - Required contract attributes are missing - Attribute values are outside valid ranges - Attribute combinations are inconsistent - Contract type is not supported

Example

>>> raise ContractValidationError(
...     "Nominal value must be positive",
...     context={"contract_id": "PAM-001", "nominal": -1000}
... )
Parameters:
Return type:

None

exception jactus.exceptions.ScheduleGenerationError(message, context=None)[source]

Bases: ActusException

Exception raised for errors during schedule generation.

This exception should be raised when: - Schedule generation algorithm fails - Date sequences are invalid - Calendar or business day adjustments fail - Cycle specifications are inconsistent

Example

>>> raise ScheduleGenerationError(
...     "Invalid cycle specification for interest payment",
...     context={"cycle": "P0M", "anchor_date": "2024-01-01"}
... )
Parameters:
Return type:

None

exception jactus.exceptions.StateTransitionError(message, context=None)[source]

Bases: ActusException

Exception raised for invalid state transitions.

This exception should be raised when: - State transition function encounters invalid inputs - Numerical instability in state calculations - State values violate invariants - JAX compilation issues in state transitions

Example

>>> raise StateTransitionError(
...     "Negative interest rate sensitivity detected",
...     context={"event": "IP", "time": "2024-06-01", "state": {...}}
... )
Parameters:
Return type:

None

exception jactus.exceptions.PayoffCalculationError(message, context=None)[source]

Bases: ActusException

Exception raised for errors in payoff calculations.

This exception should be raised when: - Payoff function produces invalid results (NaN, Inf) - Required market data is missing - Numerical overflow/underflow occurs - Business rules are violated

Example

>>> raise PayoffCalculationError(
...     "Division by zero in fee calculation",
...     context={"event": "FP", "time": "2024-03-15", "fee_basis": 0}
... )
Parameters:
Return type:

None

exception jactus.exceptions.ObserverError(message, context=None)[source]

Bases: ActusException

Exception raised for risk factor or child contract observation errors.

This exception should be raised when: - Risk factor observer cannot provide required data - Market data is unavailable for the requested time - Child contract evaluation fails - Observer state is inconsistent

Example

>>> raise ObserverError(
...     "No exchange rate data available for requested date",
...     context={"currency_pair": "EUR/USD", "date": "2024-01-15"}
... )
Parameters:
Return type:

None

exception jactus.exceptions.DateTimeError(message, context=None)[source]

Bases: ActusException

Exception raised for date/time parsing or calculation errors.

This exception should be raised when: - Date string parsing fails - Timezone conversion issues occur - Date arithmetic produces invalid results - Calendar operations fail

Example

>>> raise DateTimeError(
...     "Unable to parse ISO date string",
...     context={"date_string": "2024-13-45", "format": "ISO8601"}
... )
Parameters:
Return type:

None

exception jactus.exceptions.ConventionError(message, context=None)[source]

Bases: ActusException

Exception raised for day count or business day convention errors.

This exception should be raised when: - Unknown day count convention specified - Business day adjustment fails - End-of-month rules produce invalid dates - Convention application is ambiguous

Example

>>> raise ConventionError(
...     "Unsupported day count convention",
...     context={"convention": "ACT/999", "supported": ["30E/360", "ACT/365"]}
... )
Parameters:
Return type:

None

exception jactus.exceptions.ConfigurationError(message, context=None)[source]

Bases: ActusException

Exception raised for configuration and initialization errors.

This exception should be raised when: - Package configuration is invalid - Required environment variables are missing - Initialization parameters are inconsistent - Resource loading fails

Example

>>> raise ConfigurationError(
...     "Invalid logging configuration",
...     context={"log_level": "INVALID", "valid_levels": ["DEBUG", "INFO"]}
... )
Parameters:
Return type:

None

exception jactus.exceptions.EngineError(message, context=None)[source]

Bases: ActusException

Exception raised for simulation and portfolio engine errors.

This exception should be raised when: - Portfolio simulation fails - Batch processing encounters errors - Resource allocation issues occur - Parallel execution fails

Example

>>> raise EngineError(
...     "Portfolio simulation failed due to memory constraints",
...     context={"num_contracts": 10000, "memory_gb": 8}
... )
Parameters:
Return type:

None

Logging

Centralized logging configuration for jactus.

This module provides a flexible logging setup that can be configured via environment variables or programmatically. It supports multiple handlers, structured logging, and performance monitoring.

class jactus.logging_config.StructuredFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]

Bases: Formatter

Formatter that outputs log records as JSON for structured logging.

format(record)[source]

Format the log record as JSON.

Parameters:

record (LogRecord) – The log record to format

Returns:

JSON string representation of the log record

Return type:

str

jactus.logging_config.get_logger(name, level=None)[source]

Get a logger instance with the specified name and level.

Parameters:

  • name (str) – Name of the logger (typically __name__ of the calling module)

  • level (str | None) – Optional log level override. If not specified, uses environment variable or default level.

Returns:

Configured logger instance

Return type:

Logger

Example

>>> logger = get_logger(__name__)
>>> logger.info("Processing contract", extra={"contract_id": "PAM-001"})
jactus.logging_config.configure_logging(level=None, log_file=None, console=True, structured=False, max_bytes=10485760, backup_count=5)[source]

Configure logging for the entire jactus package.

This function sets up logging handlers and formatters for the package. It should typically be called once at application startup.

Parameters:

  • level (str | None) – Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Defaults to environment variable or INFO.

  • log_file (str | None) – Path to log file. Defaults to environment variable or jactus.log in current directory.

  • console (bool) – Whether to log to console (stdout). Default: True

  • structured (bool) – Whether to use JSON structured logging. Default: False Can be overridden by ACTUS_JAX_STRUCTURED_LOGS env var.

  • max_bytes (int) – Maximum size of log file before rotation (default: 10MB)

  • backup_count (int) – Number of backup log files to keep (default: 5)

Return type:

None

Example

>>> # Simple configuration
>>> configure_logging(level="DEBUG")
>>>
>>> # Full configuration with file logging
>>> configure_logging(
...     level="INFO",
...     log_file="/var/log/jactus.log",
...     structured=True
... )
jactus.logging_config.get_performance_logger(name)[source]

Get a logger configured for performance monitoring.

Performance loggers are typically used to log timing information, memory usage, and other performance metrics. They default to DEBUG level and can be controlled separately from regular loggers.

Parameters:

name (str) – Name of the performance logger

Returns:

Logger configured for performance monitoring

Return type:

Logger

Example

>>> perf_logger = get_performance_logger("jactus.engine")
>>> import time
>>> start = time.time()
>>> # ... do work ...
>>> elapsed = time.time() - start
>>> perf_logger.debug("Portfolio simulation completed",
...                   extra={"duration_ms": elapsed * 1000,
...                          "num_contracts": 1000})
jactus.logging_config.disable_logging()[source]

Disable all jactus logging.

This is useful for tests or applications that want to suppress all logging output from the package.

Return type:

None

Quick Reference

Contract Type Summary

Principal Contracts

  • PAM - Principal at Maturity (interest-only loans, bonds)

  • LAM - Linear Amortizer (fixed principal amortization)

  • LAX - Exotic Linear Amortizer (variable amortization schedules)

  • NAM - Negative Amortizer (negative amortization loans)

  • ANN - Annuity (level payment amortization)

  • CLM - Call Money (callable overnight lending)

  • UMP - Undefined Maturity Profile (open-ended deposits, revolving credit)

Non-Principal Contracts

  • CSH - Cash (money market accounts, escrow)

  • STK - Stock (equity positions)

  • COM - Commodity (physical commodities)

Derivative Contracts

  • FXOUT - Foreign Exchange Outright (FX forwards, swaps)

  • OPTNS - Options (calls, puts, European/American)

  • FUTUR - Futures (standardized forward contracts)

  • SWPPV - Plain Vanilla Swap (fixed vs floating interest rate swaps)

  • SWAPS - Generic Swap (cross-currency swaps, multi-leg swaps)

  • CAPFL - Cap/Floor (interest rate caps and floors)

  • CEG - Credit Enhancement Guarantee (credit protection)

  • CEC - Credit Enhancement Collateral (collateral management)

Key Classes and Functions

Creating Contracts:

from jactus.contracts import create_contract
from jactus.core import ContractAttributes, ContractType, ContractRole, ActusDateTime
from jactus.observers import ConstantRiskFactorObserver

attrs = ContractAttributes(
    contract_id="LOAN-001",
    contract_type=ContractType.PAM,
    contract_role=ContractRole.RPA,
    status_date=ActusDateTime(2024, 1, 1),
    initial_exchange_date=ActusDateTime(2024, 1, 15),
    maturity_date=ActusDateTime(2025, 1, 15),
    notional_principal=100_000.0,
    nominal_interest_rate=0.05,
)
rf_observer = ConstantRiskFactorObserver(constant_value=0.05)
contract = create_contract(attrs, rf_observer)

Running Simulations:

# Run simulation — returns SimulationHistory
result = contract.simulate()

# Access events
for event in result.events:
    print(f"{event.event_time}: {event.event_type.name} ${event.payoff:,.2f}")

Working with Time:

from jactus.core import ActusDateTime

dt = ActusDateTime(2024, 1, 1, 0, 0, 0)

Schedule Generation:

from jactus.utilities.schedules import generate_schedule

schedule = generate_schedule(
    start=start_date,
    end=end_date,
    cycle="3M"  # Quarterly
)

See Also