Source code for ergodic_insurance.manufacturer
# pylint: disable=too-many-lines
"""Widget manufacturer financial model implementation.
This module implements the core financial model for a widget manufacturing
company, providing comprehensive balance sheet management, insurance claim
processing, and stochastic modeling capabilities. It serves as the central
component of the ergodic insurance optimization framework.
The manufacturer model simulates realistic business operations including:
- Asset-based revenue generation with configurable turnover ratios
- Operating income calculations with industry-standard margins
- Multi-layer insurance claim processing with deductibles and limits
- Letter of credit collateral management for claim liabilities
- Actuarial claim payment schedules over multiple years
- Dynamic balance sheet evolution with growth and volatility
- Integration with sophisticated stochastic processes
- Comprehensive financial metrics and ratio analysis
Key Components:
- :class:`WidgetManufacturer`: Main financial model class
- :class:`ClaimLiability`: Actuarial claim payment tracking (re-exported)
- :class:`TaxHandler`: Tax calculation and accrual (re-exported)
Examples:
Basic manufacturer setup and simulation::
from ergodic_insurance import ManufacturerConfig
from ergodic_insurance.manufacturer import WidgetManufacturer
config = ManufacturerConfig(
initial_assets=10_000_000,
asset_turnover_ratio=0.8,
base_operating_margin=0.08,
tax_rate=0.25,
retention_ratio=0.7
)
manufacturer = WidgetManufacturer(config)
metrics = manufacturer.step(
letter_of_credit_rate=0.015,
growth_rate=0.05
)
print(f"ROE: {metrics['roe']:.1%}")
"""
import copy as copy_module
from dataclasses import dataclass, field
from decimal import Decimal
import logging
import random
from typing import Any, Dict, List, Optional, Union
try:
from ergodic_insurance.accrual_manager import (
AccrualManager,
AccrualType,
PaymentSchedule,
)
from ergodic_insurance.config import ManufacturerConfig
from ergodic_insurance.decimal_utils import (
ONE,
ZERO,
MetricsDict,
enable_float_mode,
to_decimal,
)
from ergodic_insurance.insurance_accounting import InsuranceAccounting
from ergodic_insurance.ledger import AccountName, Ledger, TransactionType
from ergodic_insurance.stochastic_processes import StochasticProcess
except ImportError:
try:
from .accrual_manager import AccrualManager, AccrualType, PaymentSchedule
from .config import ManufacturerConfig
from .decimal_utils import ONE, ZERO, MetricsDict, enable_float_mode, to_decimal
from .insurance_accounting import InsuranceAccounting
from .ledger import AccountName, Ledger, TransactionType
from .stochastic_processes import StochasticProcess
except ImportError:
from accrual_manager import ( # type: ignore[no-redef]
AccrualManager,
AccrualType,
PaymentSchedule,
)
from config import ManufacturerConfig # type: ignore[no-redef]
from decimal_utils import ( # type: ignore[no-redef]
ONE,
ZERO,
MetricsDict,
enable_float_mode,
to_decimal,
)
from insurance_accounting import InsuranceAccounting # type: ignore[no-redef]
from ledger import AccountName, Ledger, TransactionType # type: ignore[no-redef]
from stochastic_processes import StochasticProcess # type: ignore[no-redef]
# Re-exports for backward compatibility (Issue #305)
from ergodic_insurance.claim_liability import ( # noqa: F401 pylint: disable=ungrouped-imports
ClaimLiability,
)
# Import mixins
from ergodic_insurance.manufacturer_balance_sheet import ( # pylint: disable=ungrouped-imports
BalanceSheetMixin,
)
from ergodic_insurance.manufacturer_claims import ClaimProcessingMixin
from ergodic_insurance.manufacturer_income import IncomeCalculationMixin
from ergodic_insurance.manufacturer_metrics import MetricsCalculationMixin
from ergodic_insurance.manufacturer_solvency import SolvencyMixin
from ergodic_insurance.tax_handler import TaxHandler # noqa: F401 pylint: disable=unused-import
logger = logging.getLogger(__name__)
@dataclass
class _PPECohort:
"""A vintage-year cohort of PP&E assets for depreciation tracking (Issue #1321).
Tracks gross cost and cumulative tax depreciation per acquisition cohort
so that per-cohort remaining-basis caps are applied correctly in the DTL
calculation (ASC 740-10-25-20).
"""
amount: Decimal
tax_accumulated_depreciation: Decimal = field(default_factory=lambda: ZERO)
[docs]
class WidgetManufacturer(
BalanceSheetMixin,
ClaimProcessingMixin,
IncomeCalculationMixin,
SolvencyMixin,
MetricsCalculationMixin,
):
"""Financial model for a widget manufacturing company.
This class models the complete financial operations of a manufacturing
company including revenue generation, claim processing, collateral
management, and balance sheet evolution over time.
The manufacturer maintains a balance sheet with assets, equity, and tracks
insurance-related collateral. It can process insurance claims with multi-year
payment schedules and manages working capital requirements.
Attributes:
config: Manufacturing configuration parameters
stochastic_process: Optional stochastic process for revenue volatility
assets: Current total assets
collateral: Letter of credit collateral for insurance claims
restricted_assets: Assets restricted as collateral
equity: Current equity (assets minus liabilities)
year: Current simulation year
outstanding_liabilities: List of active claim liabilities
metrics_history: Historical metrics for each simulation period
bankruptcy: Whether the company has gone bankrupt
bankruptcy_year: Year when bankruptcy occurred (if applicable)
Example:
Running a multi-year simulation::
manufacturer = WidgetManufacturer(config)
for year in range(10):
losses, _ = loss_generator.generate_losses(duration=1, revenue=revenue)
for loss in losses:
manufacturer.process_insurance_claim(
loss.amount, deductible, limit
)
metrics = manufacturer.step(letter_of_credit_rate=0.015)
print(f"Year {year}: ROE={metrics['roe']:.1%}")
"""
def __init__(
self,
config: ManufacturerConfig,
stochastic_process: Optional[StochasticProcess] = None,
use_float: bool = False,
simulation_mode: bool = False,
):
"""Initialize manufacturer with configuration parameters.
Args:
config (ManufacturerConfig): Manufacturing configuration parameters.
stochastic_process (Optional[StochasticProcess]): Optional stochastic
process for adding revenue volatility. Defaults to None.
use_float (bool): If True, enable float mode for this instance
to avoid Decimal overhead in Monte Carlo hot paths (Issue #1142).
simulation_mode (bool): If True, the ledger only maintains balance
caches without storing individual entries (Issue #1146).
"""
self._use_float = use_float
if use_float:
enable_float_mode()
self.config = config
self.stochastic_process = stochastic_process
# Initialize the event-sourcing ledger FIRST
self.ledger = Ledger(simulation_mode=simulation_mode)
# Track original prepaid premium for amortization calculation
self._original_prepaid_premium: Decimal = to_decimal(0)
# Insurance accounting module
self.insurance_accounting = InsuranceAccounting()
# Accrual management for timing differences (Issue #277: fiscal-year-aware)
self.accrual_manager = AccrualManager(fiscal_year_end=config.fiscal_year_end)
# Operating parameters
self.asset_turnover_ratio = config.asset_turnover_ratio
self.base_operating_margin = config.base_operating_margin
self.tax_rate = config.tax_rate
self.retention_ratio = config.retention_ratio
# Tax handler with NOL carryforward tracking (Issue #365, #808)
self.tax_handler = TaxHandler(
tax_rate=config.tax_rate,
accrual_manager=self.accrual_manager,
nol_carryforward=to_decimal(0),
nol_limitation_pct=(
config.nol_limitation_pct if config.nol_carryforward_enabled else 0.0
),
apply_tcja_limitation=config.apply_tcja_limitation,
)
self._nol_carryforward_enabled = config.nol_carryforward_enabled
# Claim tracking
self.claim_liabilities: List[ClaimLiability] = []
self.current_year = 0
self.current_month = 0
# Insurance cost tracking for tax purposes
self.period_insurance_premiums: Decimal = to_decimal(0)
self.period_insurance_losses: Decimal = to_decimal(0)
self.period_insurance_lae: Decimal = to_decimal(0) # LAE per ASC 944-40 (Issue #468)
# Cached net income from step() for use by calculate_metrics() (Issue #617)
self._period_net_income: Optional[Decimal] = None
# Track actual dividends paid
self._last_dividends_paid: Decimal = to_decimal(0)
# Solvency tracking
self.is_ruined = False
self.ruin_month: Optional[int] = None
# Reserve development tracking (Issue #470, ASC 944-40-25)
self.period_adverse_development: Decimal = to_decimal(0)
self.period_favorable_development: Decimal = to_decimal(0)
self._enable_reserve_development = config.enable_reserve_development
self._reserve_rng: Optional[random.Random] = (
random.Random(42) if self._enable_reserve_development else None
)
# Metrics tracking
self.metrics_history: List[MetricsDict] = []
# Store initial values for base comparisons
self._initial_assets: Decimal = to_decimal(config.initial_assets)
self._initial_equity: Decimal = to_decimal(config.initial_assets)
# Compute initial balance sheet values
# ppe_ratio is guaranteed non-None after model_validator
assert config.ppe_ratio is not None
initial_gross_ppe: Decimal = to_decimal(config.initial_assets * config.ppe_ratio)
initial_accumulated_depreciation: Decimal = ZERO
# Current Assets - initialize working capital to steady state
initial_revenue = to_decimal(config.initial_assets * config.asset_turnover_ratio)
initial_cogs = initial_revenue * to_decimal(1 - config.base_operating_margin)
initial_accounts_receivable: Decimal = initial_revenue * to_decimal(45 / 365)
initial_inventory: Decimal = initial_cogs * to_decimal(60 / 365)
initial_prepaid_insurance: Decimal = ZERO
working_capital_assets = initial_accounts_receivable + initial_inventory
initial_cash: Decimal = (
to_decimal(config.initial_assets * (1 - config.ppe_ratio)) - working_capital_assets
)
initial_accounts_payable: Decimal = ZERO
initial_collateral: Decimal = ZERO
initial_restricted_assets: Decimal = ZERO
# Record all initial balances to ledger
self._record_initial_balances(
cash=initial_cash,
accounts_receivable=initial_accounts_receivable,
inventory=initial_inventory,
prepaid_insurance=initial_prepaid_insurance,
gross_ppe=initial_gross_ppe,
accumulated_depreciation=initial_accumulated_depreciation,
accounts_payable=initial_accounts_payable,
collateral=initial_collateral,
restricted_assets=initial_restricted_assets,
)
# PP&E vintage cohort tracking for accurate DTL calculation (Issue #1321)
self._ppe_cohorts: List[_PPECohort] = []
if initial_gross_ppe > ZERO:
self._ppe_cohorts.append(_PPECohort(amount=initial_gross_ppe))
# Pre-compute frequently used config conversions (Issue #1142)
self._cache_config_values()
def _cache_config_values(self) -> None:
"""Pre-compute to_decimal() conversions for config values (Issue #1142)."""
self._cached_capex_ratio = to_decimal(self.config.capex_to_depreciation_ratio)
self._cached_base_margin_complement = to_decimal(1 - self.base_operating_margin)
self._cached_book_life = to_decimal(self.config.ppe_useful_life_years)
def _record_initial_balances(
self,
cash: Decimal,
accounts_receivable: Decimal,
inventory: Decimal,
prepaid_insurance: Decimal,
gross_ppe: Decimal,
accumulated_depreciation: Decimal,
accounts_payable: Decimal,
collateral: Decimal,
restricted_assets: Decimal,
) -> None:
"""Record initial balance sheet entries in the ledger.
This establishes the opening balances for all accounts at year 0.
Uses equity (retained_earnings) as the balancing entry.
Args:
cash: Initial cash position
accounts_receivable: Initial accounts receivable
inventory: Initial inventory
prepaid_insurance: Initial prepaid insurance
gross_ppe: Initial gross property, plant & equipment
accumulated_depreciation: Initial accumulated depreciation
accounts_payable: Initial accounts payable
collateral: Initial letter of credit collateral
restricted_assets: Initial restricted assets
"""
if cash > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.CASH,
credit_account=AccountName.RETAINED_EARNINGS,
amount=cash,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial cash position",
)
if accounts_receivable > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.ACCOUNTS_RECEIVABLE,
credit_account=AccountName.RETAINED_EARNINGS,
amount=accounts_receivable,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial accounts receivable",
)
if inventory > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.INVENTORY,
credit_account=AccountName.RETAINED_EARNINGS,
amount=inventory,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial inventory",
)
if prepaid_insurance > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.PREPAID_INSURANCE,
credit_account=AccountName.RETAINED_EARNINGS,
amount=prepaid_insurance,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial prepaid insurance",
)
if gross_ppe > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.GROSS_PPE,
credit_account=AccountName.RETAINED_EARNINGS,
amount=gross_ppe,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial gross PP&E",
)
if accumulated_depreciation > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.RETAINED_EARNINGS,
credit_account=AccountName.ACCUMULATED_DEPRECIATION,
amount=accumulated_depreciation,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial accumulated depreciation",
)
if accounts_payable > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.RETAINED_EARNINGS,
credit_account=AccountName.ACCOUNTS_PAYABLE,
amount=accounts_payable,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial accounts payable",
)
if restricted_assets > ZERO:
self.ledger.record_double_entry(
date=0,
debit_account=AccountName.RESTRICTED_CASH,
credit_account=AccountName.RETAINED_EARNINGS,
amount=restricted_assets,
transaction_type=TransactionType.EQUITY_ISSUANCE,
description="Initial restricted assets",
)
# Properties for FinancialStateProvider protocol
@property
def current_revenue(self) -> Decimal:
"""Get current revenue based on current assets and turnover ratio."""
return self.calculate_revenue()
@property
def current_assets(self) -> Decimal:
"""Get current total assets."""
return self.total_assets
@property
def current_equity(self) -> Decimal:
"""Get current equity value."""
return self.equity
@property
def base_revenue(self) -> Decimal:
"""Get base (initial) revenue for comparison."""
return self._initial_assets * to_decimal(self.config.asset_turnover_ratio)
@property
def base_assets(self) -> Decimal:
"""Get base (initial) assets for comparison."""
return self._initial_assets
@property
def base_equity(self) -> Decimal:
"""Get base (initial) equity for comparison."""
return self._initial_equity
# ========================================================================
# Serialization support
# ========================================================================
[docs]
def __deepcopy__(self, memo: Dict[int, Any]) -> "WidgetManufacturer":
"""Create a deep copy preserving all state for Monte Carlo forking.
Args:
memo: Dictionary of already copied objects (for cycle detection)
Returns:
Independent copy of this WidgetManufacturer with all state preserved
"""
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
for key, value in self.__dict__.items():
setattr(result, key, copy_module.deepcopy(value, memo))
return result
[docs]
def __getstate__(self) -> Dict[str, Any]:
"""Get state for pickling (required for Windows multiprocessing).
Returns:
Dictionary of all instance attributes
"""
return self.__dict__.copy()
[docs]
def __setstate__(self, state: Dict[str, Any]) -> None:
"""Restore state from pickle (required for Windows multiprocessing).
Args:
state: Dictionary of instance attributes to restore
"""
self.__dict__.update(state)
# ========================================================================
# Accrual coordination
# ========================================================================
[docs]
def process_accrued_payments(
self,
time_resolution: str = "annual",
max_payable: Optional[Union[Decimal, float]] = None,
) -> Decimal:
"""Process due accrual payments for the current period.
Args:
time_resolution: "annual" or "monthly" for determining current period
max_payable: Optional maximum amount that can be paid.
Returns:
Total cash payments made for accruals in this period
"""
if time_resolution == "monthly":
period = self.current_year * 12 + self.current_month
else:
period = self.current_year * 12
self.accrual_manager.current_period = period
payments_due = self.accrual_manager.get_payments_due(period)
total_due = sum((to_decimal(v) for v in payments_due.values()), to_decimal(0))
if max_payable is not None:
max_total_payable: Decimal = min(total_due, to_decimal(max_payable))
else:
current_equity = self.equity
max_total_payable = (
min(total_due, current_equity) if current_equity > ZERO else to_decimal(0)
)
if total_due > max_total_payable:
logger.warning(
f"LIMITED LIABILITY: Capping total accrued payments. "
f"Due: ${total_due:,.2f}, Payable: ${max_total_payable:,.2f}"
)
payment_ratio: Decimal = (
max_total_payable / total_due if total_due > ZERO else to_decimal(0)
)
total_paid: Decimal = to_decimal(0)
for accrual_type, amount_due in payments_due.items():
payable_amount: Decimal = to_decimal(amount_due) * payment_ratio
unpayable_amount = to_decimal(amount_due) - payable_amount
if payable_amount > ZERO:
self.accrual_manager.process_payment(accrual_type, payable_amount, period)
total_paid += payable_amount
if accrual_type == AccrualType.TAXES:
trans_type = TransactionType.TAX_PAYMENT
debit_account = AccountName.ACCRUED_TAXES
elif accrual_type == AccrualType.WAGES:
trans_type = TransactionType.WAGE_PAYMENT
debit_account = AccountName.ACCRUED_WAGES
elif accrual_type == AccrualType.INTEREST:
trans_type = TransactionType.INTEREST_PAYMENT
debit_account = AccountName.ACCRUED_INTEREST
else:
trans_type = TransactionType.PAYMENT
debit_account = AccountName.ACCRUED_EXPENSES
self.ledger.record_double_entry(
date=self.current_year,
debit_account=debit_account,
credit_account=AccountName.CASH,
amount=payable_amount,
transaction_type=trans_type,
description=f"Accrued {accrual_type.value} payment",
month=self.current_month,
)
logger.debug(f"Paid accrued {accrual_type.value}: ${payable_amount:,.2f}")
if unpayable_amount > ZERO:
# Discharge unpayable amount from AccrualManager (Issue #1063)
self.accrual_manager.process_payment(accrual_type, unpayable_amount, period)
# Reverse phantom liability from ledger per ASC 405-20 (Issue #1063)
# Dr ACCRUED_XXX (reduce liability), Cr RETAINED_EARNINGS (equity absorbs)
if accrual_type == AccrualType.TAXES:
discharge_account = AccountName.ACCRUED_TAXES
elif accrual_type == AccrualType.WAGES:
discharge_account = AccountName.ACCRUED_WAGES
elif accrual_type == AccrualType.INTEREST:
discharge_account = AccountName.ACCRUED_INTEREST
else:
discharge_account = AccountName.ACCRUED_EXPENSES
self.ledger.record_double_entry(
date=self.current_year,
debit_account=discharge_account,
credit_account=AccountName.RETAINED_EARNINGS,
amount=unpayable_amount,
transaction_type=TransactionType.WRITE_OFF,
description=f"Liability discharge: unpayable {accrual_type.value} (ASC 405-20)",
month=self.current_month,
)
logger.warning(
f"LIMITED LIABILITY: Discharged ${unpayable_amount:,.2f} of unpayable {accrual_type.value} from liabilities"
)
if total_paid > ZERO:
logger.info(f"Total accrual payments this period: ${total_paid:,.2f}")
return total_paid
[docs]
def record_wage_accrual(
self,
amount: float,
payment_schedule: PaymentSchedule = PaymentSchedule.IMMEDIATE,
) -> None:
"""Record accrued wages to be paid later.
Args:
amount: Wage amount to accrue
payment_schedule: When wages will be paid
"""
self.accrual_manager.record_expense_accrual(
item_type=AccrualType.WAGES,
amount=amount,
payment_schedule=payment_schedule,
description=f"Period {self.current_year} wages",
)
# ========================================================================
# Simulation orchestration
# ========================================================================
def _handle_insolvent_step(self, time_resolution: str) -> MetricsDict:
"""Handle a simulation step when the company is already insolvent.
Args:
time_resolution: "annual" or "monthly" for simulation step.
Returns:
Dictionary of metrics for this time step.
"""
logger.warning("Company is already insolvent, skipping step")
metrics = self.calculate_metrics()
metrics["year"] = self.current_year
metrics["month"] = self.current_month if time_resolution == "monthly" else 0
self._increment_time(time_resolution)
return metrics
def _increment_time(self, time_resolution: str) -> None:
"""Increment the current time based on resolution.
Args:
time_resolution: "annual" or "monthly" for simulation step.
"""
if time_resolution == "monthly":
self.current_month += 1
if self.current_month >= 12:
self.current_month = 0
self.current_year += 1
else:
self.current_year += 1
[docs]
def step(
self,
letter_of_credit_rate: Union[Decimal, float] = 0.015,
growth_rate: Union[Decimal, float] = 0.0,
time_resolution: str = "annual",
apply_stochastic: bool = False,
) -> MetricsDict:
"""Execute one time step of the financial model simulation.
Args:
letter_of_credit_rate (float): Annual interest rate for letter of credit.
growth_rate (float): Revenue growth rate for the period.
time_resolution (str): "annual" or "monthly".
apply_stochastic (bool): Whether to apply stochastic shocks.
Returns:
Dict[str, float]: Comprehensive financial metrics dictionary.
"""
# Check if already ruined
if self.is_ruined:
return self._handle_insolvent_step(time_resolution)
# Check for potential mid-year insolvency (Issue #279)
if not self.check_liquidity_constraints(time_resolution):
return self._handle_insolvent_step(time_resolution)
# Store initial revenue for working capital calculation in monthly mode
if time_resolution == "monthly" and self.current_month == 0:
self._annual_revenue_for_wc = self.calculate_revenue(apply_stochastic)
# Calculate financial performance
revenue = self.calculate_revenue(apply_stochastic)
# Issue #1308: Record revenue BEFORE working capital adjustments.
# Revenue recognition (ASC 606) logically precedes working capital
# effects. Posting Dr AR / Cr REVENUE first ensures the WC module
# reads the correct AR balance (including this period's revenue) and
# eliminates artificial negative-cash states that triggered spurious
# working-capital facility draws.
if time_resolution == "monthly":
_revenue_to_record = revenue / to_decimal(12)
else:
_revenue_to_record = revenue
if _revenue_to_record > ZERO:
self.ledger.record_double_entry(
date=self.current_year,
debit_account=AccountName.ACCOUNTS_RECEIVABLE,
credit_account=AccountName.SALES_REVENUE,
amount=_revenue_to_record,
transaction_type=TransactionType.REVENUE,
description=f"Year {self.current_year} revenue recognition (ASC 606)",
month=self.current_month,
)
# Calculate working capital components BEFORE payment coordination.
# Issue #1302 / #1308: Revenue is already posted to AR above, so
# pass period_revenue=ZERO — the WC module reads the updated AR
# balance directly and computes collections = current_AR − target_AR.
if time_resolution == "annual":
self.calculate_working_capital_components(revenue, period_revenue=ZERO)
elif time_resolution == "monthly":
if hasattr(self, "_annual_revenue_for_wc"):
self.calculate_working_capital_components(
self._annual_revenue_for_wc, period_revenue=ZERO
)
else:
annual_revenue = self.total_assets * to_decimal(self.asset_turnover_ratio)
self.calculate_working_capital_components(annual_revenue, period_revenue=ZERO)
# COORDINATED LIMITED LIABILITY ENFORCEMENT
if time_resolution == "monthly":
period = self.current_year * 12 + self.current_month
else:
period = self.current_year * 12
# Calculate total accrual payments due
self.accrual_manager.current_period = period
accrual_payments_due = self.accrual_manager.get_payments_due(period)
total_accrual_due: Decimal = sum(
(to_decimal(v) for v in accrual_payments_due.values()), to_decimal(0)
)
# Calculate total claim payments scheduled
total_claim_due: Decimal = to_decimal(0)
if time_resolution == "annual" or self.current_month == 0:
for claim_item in self.claim_liabilities:
years_since = self.current_year - claim_item.year_incurred
scheduled_payment = claim_item.get_payment(years_since)
total_claim_due += scheduled_payment
# Cap TOTAL payments at available liquid resources
# Issue #1337: Include working capital facility in available liquidity
# so payments can draw on the credit facility up to its limit.
total_payments_due = total_accrual_due + total_claim_due
facility_limit = getattr(self.config, "working_capital_facility_limit", None)
available_liquidity = self.cash + self.restricted_assets
if facility_limit is not None:
available_liquidity += to_decimal(facility_limit)
max_total_payable: Decimal = (
min(total_payments_due, available_liquidity)
if available_liquidity > ZERO
else to_decimal(0)
)
# Allocate capped amount proportionally
if total_payments_due > ZERO:
allocation_ratio = max_total_payable / total_payments_due
max_accrual_payable: Decimal = total_accrual_due * allocation_ratio
max_claim_payable: Decimal = total_claim_due * allocation_ratio
else:
max_accrual_payable = to_decimal(0)
max_claim_payable = to_decimal(0)
if total_payments_due > max_total_payable:
logger.warning(
f"LIQUIDITY CONSTRAINT: Total payments due ${total_payments_due:,.2f} "
f"exceeds available liquidity ${available_liquidity:,.2f} "
f"(cash: ${self.cash:,.2f}, restricted: ${self.restricted_assets:,.2f}"
f"{f', facility: ${to_decimal(facility_limit):,.2f}' if facility_limit is not None else ''}). "
f"Capping at ${max_total_payable:,.2f} "
f"(Accruals: ${max_accrual_payable:,.2f}, Claims: ${max_claim_payable:,.2f})"
)
# Process accrual payments with coordinated cap
self.process_accrued_payments(time_resolution, max_payable=max_accrual_payable)
# Pay scheduled claim liabilities with coordinated cap
if time_resolution == "annual" or self.current_month == 0:
self.pay_claim_liabilities(max_payable=max_claim_payable)
# Post-payment liquidity check (Issue #1337, ASC 205-40-50-12)
# After claim/accrual payments, verify the working capital facility
# has not been breached. This catches genuine liquidity crises from
# large mandatory payments before the period continues.
if facility_limit is not None:
facility_limit_d = to_decimal(facility_limit)
if self.cash < -facility_limit_d:
logger.warning(
f"LIQUIDITY CRISIS: After claim/accrual payments, cash "
f"${self.cash:,.2f} breaches facility limit "
f"${facility_limit_d:,.2f} (Issue #1337)."
)
self.is_ruined = True
return self._handle_insolvent_step(time_resolution)
# Re-estimate reserves per ASC 944-40-25 (Issue #470)
if time_resolution == "annual" or self.current_month == 0:
self.re_estimate_reserves()
# Calculate depreciation expense for the period (Issue #1321: parameterized)
book_life = self._cached_book_life
if time_resolution == "annual":
depreciation_expense = self.record_depreciation(useful_life_years=book_life)
elif time_resolution == "monthly":
depreciation_expense = self.record_depreciation(useful_life_years=book_life * 12)
else:
depreciation_expense = to_decimal(0)
# Record capital expenditure (reinvestment in PP&E) (Issue #543)
# Capex is capitalized (Dr GROSS_PPE, Cr CASH) — does not affect income.
capex_ratio = self._cached_capex_ratio
if capex_ratio > ZERO and depreciation_expense > ZERO:
capex_amount = depreciation_expense * capex_ratio
self.record_capex(capex_amount)
# Record deferred tax liability from depreciation timing differences (Issue #367)
# When tax_depreciation_life_years < book life, accelerated tax depreciation
# creates a temporary difference that generates a DTL per ASC 740.
self._record_dtl_from_depreciation(time_resolution)
# Calculate operating income (depreciation already embedded in COGS/SGA ratios)
operating_income = self.calculate_operating_income(revenue)
# Calculate collateral costs
if time_resolution == "monthly":
collateral_costs = self.calculate_collateral_costs(letter_of_credit_rate, "monthly")
revenue = revenue / 12
operating_income = operating_income / 12
else:
collateral_costs = self.calculate_collateral_costs(letter_of_credit_rate, "annual")
# Issue #1326: Record COGS and OPEX as explicit ledger entries
# so the ledger can produce: Revenue - COGS - OPEX - Depreciation = Operating Income.
# Only the cash-consuming portions are recorded here; depreciation is
# already recorded via Dr DEPRECIATION_EXPENSE / Cr ACCUMULATED_DEPRECIATION.
cogs_expense = to_decimal(0)
opex_expense = to_decimal(0)
if revenue > ZERO:
expense_ratios = getattr(self.config, "expense_ratios", None)
if expense_ratios is not None:
cogs_ratio = to_decimal(expense_ratios.cogs_ratio)
sga_ratio = to_decimal(expense_ratios.sga_expense_ratio)
mfg_dep_alloc = to_decimal(expense_ratios.manufacturing_depreciation_allocation)
admin_dep_alloc = to_decimal(expense_ratios.admin_depreciation_allocation)
else:
cogs_ratio = to_decimal(0.85)
sga_ratio = to_decimal(0.07)
mfg_dep_alloc = to_decimal(0.7)
admin_dep_alloc = to_decimal(0.3)
# Cash-consuming portions (depreciation already recorded separately)
cogs_expense = max(ZERO, revenue * cogs_ratio - depreciation_expense * mfg_dep_alloc)
opex_expense = max(ZERO, revenue * sga_ratio - depreciation_expense * admin_dep_alloc)
if cogs_expense > ZERO:
self.ledger.record_double_entry(
date=self.current_year,
debit_account=AccountName.COST_OF_GOODS_SOLD,
credit_account=AccountName.CASH,
amount=cogs_expense,
transaction_type=TransactionType.EXPENSE,
description=f"Year {self.current_year} cost of goods sold",
month=self.current_month,
)
if opex_expense > ZERO:
self.ledger.record_double_entry(
date=self.current_year,
debit_account=AccountName.OPERATING_EXPENSES,
credit_account=AccountName.CASH,
amount=opex_expense,
transaction_type=TransactionType.EXPENSE,
description=f"Year {self.current_year} operating expenses (SGA)",
month=self.current_month,
)
# Calculate net income
net_income = self.calculate_net_income(
operating_income,
collateral_costs,
use_accrual=True,
time_resolution=time_resolution,
)
# Cache net income for calculate_metrics() to avoid double tax mutation (Issue #617)
self._period_net_income = net_income
# Update balance sheet with retained earnings (Issue #803/#1213: closing
# entries close income statement accounts to RE using net_income directly,
# avoiding the period_cash_expenses decomposition that double-counted
# depreciation embedded in base_operating_margin)
self.update_balance_sheet(
net_income,
growth_rate,
depreciation_expense,
period_revenue=revenue,
cogs_expense=cogs_expense,
opex_expense=opex_expense,
)
# Amortize prepaid insurance if applicable
if time_resolution == "monthly":
self.amortize_prepaid_insurance(months=1)
# Apply revenue growth
self._apply_growth(growth_rate, time_resolution, apply_stochastic)
# Check solvency
self.check_solvency()
# Verify accounting equation (Issue #319)
self._verify_accounting_equation()
# Calculate and store metrics
metrics = self.calculate_metrics(
period_revenue=revenue, letter_of_credit_rate=letter_of_credit_rate
)
metrics["year"] = self.current_year
metrics["month"] = self.current_month if time_resolution == "monthly" else 0
self.metrics_history.append(metrics)
# Increment time
self._increment_time(time_resolution)
# Reset period insurance costs for next period
self.reset_period_insurance_costs()
return metrics
def _record_dtl_from_depreciation(self, time_resolution: str) -> None:
"""Record deferred tax liability from book-tax depreciation timing difference.
When tax depreciation is accelerated relative to book depreciation
(tax_depreciation_life_years < ppe_useful_life_years), the cumulative
timing difference creates a DTL per ASC 740. With ongoing capex, new
timing differences are perpetually created, making the DTL persistent
rather than transient (Issue #367).
Uses per-cohort vintage-year tax depreciation so that fully-depreciated
cohorts contribute zero and per-cohort remaining-basis caps are applied
correctly (Issue #1321, ASC 740-10-25-20).
Args:
time_resolution: "annual" or "monthly".
"""
tax_life_cfg = self.config.tax_depreciation_life_years
if tax_life_cfg is None:
return # No accelerated depreciation configured
if time_resolution == "monthly":
tax_life = tax_life_cfg * 12
else:
tax_life = tax_life_cfg
tax_life_decimal = to_decimal(tax_life)
if self.gross_ppe <= ZERO or tax_life_decimal <= ZERO:
return
# Compute tax depreciation per vintage cohort (Issue #1321)
# Each cohort is capped at its own remaining tax basis, so
# fully-depreciated cohorts contribute zero.
total_tax_depr = ZERO
for cohort in self._ppe_cohorts:
cohort_tax_depr = cohort.amount / tax_life_decimal
remaining_basis = cohort.amount - cohort.tax_accumulated_depreciation
if remaining_basis <= ZERO:
continue
cohort_tax_depr = min(cohort_tax_depr, remaining_basis)
cohort.tax_accumulated_depreciation += cohort_tax_depr
total_tax_depr += cohort_tax_depr
# Keep TaxHandler aggregate in sync for external consumers
self.tax_handler.tax_accumulated_depreciation = sum(
(c.tax_accumulated_depreciation for c in self._ppe_cohorts), ZERO
)
# Compute desired DTL = (tax_accum - book_accum) * tax_rate
# Positive when tax depreciation is ahead of book depreciation
temp_diff = self.tax_handler.tax_accumulated_depreciation - self.accumulated_depreciation
desired_dtl = max(to_decimal(0), temp_diff * to_decimal(self.config.tax_rate))
current_dtl = self.ledger.get_balance(AccountName.DEFERRED_TAX_LIABILITY)
dtl_change = desired_dtl - current_dtl
if dtl_change > ZERO:
# DTL increased: Dr TAX_EXPENSE, Cr DEFERRED_TAX_LIABILITY
self.ledger.record_double_entry(
date=self.current_year,
debit_account=AccountName.TAX_EXPENSE,
credit_account=AccountName.DEFERRED_TAX_LIABILITY,
amount=dtl_change,
transaction_type=TransactionType.DTL_ADJUSTMENT,
description=f"Year {self.current_year} DTL recognition from depreciation timing",
month=self.current_month,
)
elif dtl_change < ZERO:
# DTL decreased (reversal): Dr DEFERRED_TAX_LIABILITY, Cr TAX_EXPENSE
self.ledger.record_double_entry(
date=self.current_year,
debit_account=AccountName.DEFERRED_TAX_LIABILITY,
credit_account=AccountName.TAX_EXPENSE,
amount=abs(dtl_change),
transaction_type=TransactionType.DTL_ADJUSTMENT,
description=f"Year {self.current_year} DTL reversal from depreciation timing",
month=self.current_month,
)
[docs]
def reset(self) -> None:
"""Reset the manufacturer to initial state for new simulation.
This method restores all financial parameters to their configured
initial values and clears historical data, enabling fresh simulation
runs from the same starting point.
Bug Fixes (Issue #305):
- FIX 1: Uses config.ppe_ratio directly instead of recalculating from margins
- FIX 2: Initializes AR/Inventory to steady-state (matching __init__) instead of zero
"""
# Reset operating parameters
self.asset_turnover_ratio = self.config.asset_turnover_ratio
self.claim_liabilities = []
self.current_year = 0
self.current_month = 0
self.is_ruined = False
self.ruin_month = None
self.metrics_history = []
# Reset period insurance cost tracking
self.period_insurance_premiums = to_decimal(0)
self.period_insurance_losses = to_decimal(0)
self.period_insurance_lae = to_decimal(0)
# Reset reserve development tracking (Issue #470)
self.period_adverse_development = to_decimal(0)
self.period_favorable_development = to_decimal(0)
if self._enable_reserve_development and self._reserve_rng is not None:
self._reserve_rng.seed(42)
# Reset dividend tracking
self._last_dividends_paid = to_decimal(0)
# Reset cached net income (Issue #617)
self._period_net_income = None
# Reset initial values (for exposure bases)
initial_assets = to_decimal(self.config.initial_assets)
self._initial_assets = initial_assets
self._initial_equity = initial_assets
# Reset accrual manager (Issue #277: fiscal-year-aware)
self.accrual_manager = AccrualManager(fiscal_year_end=self.config.fiscal_year_end)
# Reset tax handler with fresh NOL state (Issue #365)
self.tax_handler = TaxHandler(
tax_rate=self.config.tax_rate,
accrual_manager=self.accrual_manager,
nol_carryforward=to_decimal(0),
nol_limitation_pct=(
self.config.nol_limitation_pct if self._nol_carryforward_enabled else 0.0
),
)
# Track original prepaid premium for amortization calculation
self._original_prepaid_premium = to_decimal(0)
# FIX 1 (Issue #305): Use config.ppe_ratio directly, same as __init__()
# Previously, reset() recalculated ppe_ratio from margin thresholds,
# which could diverge from the config value used in __init__().
# Type ignore: ppe_ratio is guaranteed non-None after model_validator
ppe_ratio = to_decimal(self.config.ppe_ratio)
initial_gross_ppe: Decimal = initial_assets * ppe_ratio
initial_accumulated_depreciation: Decimal = ZERO
# FIX 2 (Issue #305): Initialize AR/Inventory to steady-state, same as __init__()
# Previously, reset() set AR/Inventory to zero, causing Year 1 "warm-up" distortion
initial_revenue = to_decimal(self.config.initial_assets * self.config.asset_turnover_ratio)
initial_cogs = initial_revenue * to_decimal(1 - self.config.base_operating_margin)
initial_accounts_receivable: Decimal = initial_revenue * to_decimal(45 / 365)
initial_inventory: Decimal = initial_cogs * to_decimal(60 / 365)
initial_prepaid_insurance: Decimal = ZERO
initial_accounts_payable: Decimal = ZERO
initial_collateral: Decimal = ZERO
initial_restricted_assets: Decimal = ZERO
# Adjust cash to fund working capital assets (same as __init__)
working_capital_assets = initial_accounts_receivable + initial_inventory
initial_cash: Decimal = initial_assets * (ONE - ppe_ratio) - working_capital_assets
# Reset ledger FIRST (single source of truth)
self.ledger = Ledger(simulation_mode=self.ledger._simulation_mode)
self._record_initial_balances(
cash=initial_cash,
accounts_receivable=initial_accounts_receivable,
inventory=initial_inventory,
prepaid_insurance=initial_prepaid_insurance,
gross_ppe=initial_gross_ppe,
accumulated_depreciation=initial_accumulated_depreciation,
accounts_payable=initial_accounts_payable,
collateral=initial_collateral,
restricted_assets=initial_restricted_assets,
)
# Reset PP&E vintage cohort tracking (Issue #1321)
self._ppe_cohorts = []
if initial_gross_ppe > ZERO:
self._ppe_cohorts.append(_PPECohort(amount=initial_gross_ppe))
# Reset stochastic process if present
if self.stochastic_process is not None:
self.stochastic_process.reset()
logger.info("Manufacturer reset to initial state")
[docs]
def copy(self) -> "WidgetManufacturer":
"""Create a deep copy of the manufacturer for parallel simulations.
Returns:
WidgetManufacturer: A new manufacturer instance with same configuration.
"""
new_manufacturer = WidgetManufacturer(
config=self.config,
stochastic_process=(
copy_module.deepcopy(self.stochastic_process) if self.stochastic_process else None
),
)
logger.debug("Created copy of manufacturer")
return new_manufacturer
[docs]
@classmethod
def create_fresh(
cls,
config: ManufacturerConfig,
stochastic_process: Optional[StochasticProcess] = None,
use_float: bool = False,
simulation_mode: bool = False,
) -> "WidgetManufacturer":
"""Create a fresh manufacturer from configuration alone.
Factory method that avoids ``copy.deepcopy`` by constructing a new
instance directly from its config. Use this in hot loops (e.g. Monte
Carlo workers) where each simulation needs a clean initial state.
Args:
config: Manufacturing configuration parameters.
stochastic_process: Optional stochastic process instance. The
caller is responsible for ensuring independence (e.g. by
deep-copying the process once before passing it in).
use_float: If True, enable float mode (Issue #1142).
simulation_mode: If True, ledger skips entry storage (Issue #1146).
Returns:
A new WidgetManufacturer in its initial state.
"""
return cls(
config=config,
stochastic_process=stochastic_process,
use_float=use_float,
simulation_mode=simulation_mode,
)