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

Factory Function

Principal Contracts

PAM - Principal at Maturity

LAM - Linear Amortizer

LAX - Exotic Linear Amortizer

NAM - Negative Amortizer

ANN - Annuity

CLM - Call Money

UMP - Undefined Maturity Profile

Non-Principal Contracts

CSH - Cash

STK - Stock

COM - Commodity

Derivative Contracts

FXOUT - Foreign Exchange Outright

OPTNS - Options

FUTUR - Futures

SWPPV - Plain Vanilla Swap

SWAPS - Generic Swap

CAPFL - Cap/Floor

CEG - Credit Enhancement Guarantee

CEC - Credit Enhancement Collateral

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:

  • contract (BaseContract) – Contract instance (must have get_events() and payoff_function)

  • time (ActusDateTime) – Valuation time

  • risk_factor_observer (RiskFactorObserver) – Observer for risk factors

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

Simulation Engine

CLI

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