AnalysisLayer
The Analysis Layer provides a modular system for interpreting ACTUS contract simulation results.
It builds on a shared base class Analysis and offers specialized subclasses for common use cases:
IncomeAnalysis: tracks earnings from interest and feesLiquidityAnalysis: tracks total net cash movementValueAnalysis: computes nominal and discounted value (NPV)
Each analysis operates on a CashFlowStream and can be customized by time range or aggregation frequency.
🧱 Base Class: Analysis
All analysis classes inherit from Analysis, which provides access to:
- The original portfolio (
.portfolio) - The cash flow events (
.events_df) - Risk factor data (
.risk_factors)
class Analysis(ABC):
def __init__(self, cf_stream):
self.cf_stream = cf_stream
self.portfolio = cf_stream.portfolio
self.risk_factors = cf_stream.riskFactors
self.events_df = cf_stream.events_df.copy()
⏱ Frequency Options & Time Filtering
Most analyses support aggregation and filtering:
| Code | Description |
|---|---|
| M | Monthly |
| Q | Quarterly |
| 2Q | Semi-Annually |
| Y | Yearly |
Optional parameters start and end define the analysis window.
IncomeAnalysis(cf_stream, freq="Q", start="2025-01-01", end="2027-12-31")
📈 IncomeAnalysis
Captures interest (IP) and fee payments (FP) over time.
ia = IncomeAnalysis(cf_stream, freq="M")
print(ia.results)
ia.plot()
- Output: net income per time bucket
- Supports
.plot()andreturn_fig=True
💧 LiquidityAnalysis
Summarizes total net cash (incoming and outgoing) across all events:
liq = LiquidityAnalysis(cf_stream, freq="Y")
print(liq.results)
liq.plot()
💰 ValueAnalysis
Computes:
- Undiscounted nominal value of all future payoffs
- Net Present Value (NPV) using a discount curve or flat rate
val = ValueAnalysis(cf_stream, as_of_date="2025-01-01", discount_curve_code="USD_DISCOUNT")
print(val.summarize())
If multiple curves are provided, the user must specify one using discount_curve_code.
If none are available, a flat_rate=0.03 can be passed.
🔧 Custom Analysis
To implement your own logic:
class MyAnalysis(Analysis):
def __init__(self, cf_stream):
super().__init__(cf_stream)
self.results = self.analyze()
def analyze(self):
df = self.events_df
# your custom logic
return ...
def plot(self):
...
You can then use .results, .plot() or .to_dataframe().
🧠 Design Philosophy
Each analysis is:
- Independent: uses only what's in
CashFlowStream - Extensible: override
.analyze()and.plot()as needed - Composable: results can be combined, exported, visualized