Economic Sensitivity Analyzer Skill
name: economic-sensitivity-analyzer description: Perform advanced economic sensitivity analysis for oil & gas investments. Use for spider diagrams, 2D sensitivity surfaces, breakeven analysis, scenario comparison matrices, and decision tree analysis. Complements npv-analyzer with visualization-focused multi-variable sensitivity tools. command: /economic-sensitivity-analyzer version: 1.0.0 author: worldenergydata tags: - economics - sensitivity - analysis - visualization - decision-support
Overview
The Economic Sensitivity Analyzer provides advanced visualization and analysis tools for understanding how economic parameters affect project NPV. While the npv-analyzer provides basic sensitivity analysis, this skill focuses on multi-dimensional sensitivity visualization, breakeven analysis, and decision support tools.
When to Use
Use this skill when you need to:
- Create spider diagrams showing multi-parameter sensitivity
- Generate 2D sensitivity surfaces (contour plots) for two-variable interactions
- Calculate and visualize breakeven prices (oil, gas, or combined)
- Build scenario comparison matrices for management presentations
- Perform decision tree analysis for staged investments
- Create executive-ready sensitivity dashboards
Core Classes
SensitivityParameter
# ABOUTME: Defines a parameter for sensitivity analysis with range and display settings
# ABOUTME: Supports linear and percentage-based parameter variations
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from enum import Enum
import numpy as np
class ParameterType(Enum):
"""Type of parameter variation."""
ABSOLUTE = "absolute" # Vary by absolute values (e.g., $50, $60, $70)
PERCENTAGE = "percentage" # Vary by percentage (-30%, -20%, ..., +30%)
MULTIPLIER = "multiplier" # Vary by multiplier (0.7, 0.8, ..., 1.3)
@dataclass
class SensitivityParameter:
"""
Configuration for a sensitivity analysis parameter.
Attributes:
name: Parameter identifier (e.g., "oil_price", "capex")
display_name: Human-readable name for charts
base_value: Base case value
param_type: How to vary the parameter
variations: List of variation points
unit: Display unit (e.g., "$/bbl", "MM$")
color: Chart color for this parameter
"""
name: str
display_name: str
base_value: float
param_type: ParameterType = ParameterType.PERCENTAGE
variations: List[float] = field(default_factory=lambda: [-30, -20, -10, 0, 10, 20, 30])
unit: str = ""
color: str = "#1f77b4"
def get_values(self) -> List[float]:
"""Generate actual parameter values from variations."""
if self.param_type == ParameterType.ABSOLUTE:
return self.variations
elif self.param_type == ParameterType.PERCENTAGE:
return [self.base_value * (1 + v/100) for v in self.variations]
elif self.param_type == ParameterType.MULTIPLIER:
return [self.base_value * v for v in self.variations]
return self.variations
def get_labels(self) -> List[str]:
"""Generate display labels for chart axes."""
if self.param_type == ParameterType.PERCENTAGE:
return [f"{v:+.0f}%" for v in self.variations]
elif self.param_type == ParameterType.MULTIPLIER:
return [f"{v:.1f}x" for v in self.variations]
else:
return [f"{v:.1f}" for v in self.variations]
SpiderDiagramAnalyzer
# ABOUTME: Creates spider (radar) diagrams showing multi-parameter sensitivity
# ABOUTME: Normalizes NPV changes to percentage basis for fair comparison
from dataclasses import dataclass
from typing import Dict, List, Tuple, Optional, Callable
import numpy as np
import plotly.graph_objects as go
from plotly.subplots import make_subplots
@dataclass
class SpiderResult:
"""Results from spider diagram analysis."""
parameters: List[str]
variations: List[float] # Common variation percentages
npv_changes: Dict[str, List[float]] # param_name -> [npv_pct_change, ...]
base_npv: float
most_sensitive: str
least_sensitive: str
class SpiderDiagramAnalyzer:
"""
Generates spider (radar) diagrams for multi-parameter sensitivity analysis.
Spider diagrams show how NPV changes when each parameter is varied
by the same percentage, allowing direct comparison of parameter sensitivity.
Usage:
analyzer = SpiderDiagramAnalyzer(npv_calculator)
result = analyzer.analyze(parameters)
fig = analyzer.create_spider_chart(result)
"""
def __init__(self, npv_calculator: Callable[[Dict[str, float]], float]):
"""
Initialize analyzer with NPV calculation function.
Args:
npv_calculator: Function that takes parameter dict and returns NPV
"""
self.npv_calculator = npv_calculator
def analyze(
self,
parameters: List[SensitivityParameter],
variations: List[float] = None
) -> SpiderResult:
"""
Perform spider diagram sensitivity analysis.
Args:
parameters: List of parameters to analyze
variations: Percentage variations to test (default: -30 to +30)
Returns:
SpiderResult with NPV changes for each parameter
"""
if variations is None:
variations = [-30, -20, -10, 0, 10, 20, 30]
# Calculate base case NPV
base_params = {p.name: p.base_value for p in parameters}
base_npv = self.npv_calculator(base_params)
# Calculate NPV for each parameter variation
npv_changes = {}
max_sensitivity = 0
min_sensitivity = float('inf')
most_sensitive = ""
least_sensitive = ""
for param in parameters:
changes = []
for var_pct in variations:
# Vary only this parameter
test_params = base_params.copy()
test_params[param.name] = param.base_value * (1 + var_pct / 100)
# Calculate NPV and percentage change from base
test_npv = self.npv_calculator(test_params)
pct_change = ((test_npv - base_npv) / abs(base_npv)) * 100 if base_npv != 0 else 0
changes.append(pct_change)
npv_changes[param.name] = changes
# Track most/least sensitive (use range of NPV changes)
sensitivity = max(changes) - min(changes)
if sensitivity > max_sensitivity:
max_sensitivity = sensitivity
most_sensitive = param.name
if sensitivity < min_sensitivity:
min_sensitivity = sensitivity
least_sensitive = param.name
return SpiderResult(
parameters=[p.name for p in parameters],
variations=variations,
npv_changes=npv_changes,
base_npv=base_npv,
most_sensitive=most_sensitive,
least_sensitive=least_sensitive
)
def create_spider_chart(
self,
result: SpiderResult,
title: str = "NPV Sensitivity Spider Diagram",
show_legend: bool = True
) -> go.Figure:
"""
Create interactive spider/radar chart.
Args:
result: SpiderResult from analyze()
title: Chart title
show_legend: Whether to show parameter legend
Returns:
Plotly Figure object
"""
fig = go.Figure()
# Color palette for parameters
colors = [
'#1f77b4', '#ff7f0e', '#2ca02c', '#d62728',
'#9467bd', '#8c564b', '#e377c2', '#7f7f7f'
]
for i, param in enumerate(result.parameters):
changes = result.npv_changes[param]
# Close the spider by repeating first point
r_values = changes + [changes[0]]
theta_values = [f"{v:+.0f}%" for v in result.variations] + [f"{result.variations[0]:+.0f}%"]
fig.add_trace(go.Scatterpolar(
r=r_values,
theta=theta_values,
fill='toself',
fillcolor=f'rgba{tuple(list(int(colors[i % len(colors)].lstrip("#")[j:j+2], 16) for j in (0, 2, 4)) + [0.1])}',
line=dict(color=colors[i % len(colors)], width=2),
name=param.replace("_", " ").title()
))
fig.update_layout(
polar=dict(
radialaxis=dict(
visible=True,
title="NPV Change (%)",
ticksuffix="%"
),
angularaxis=dict(
direction="clockwise",
rotation=90
)
),
showlegend=show_legend,
title=dict(text=title, x=0.5),
height=600,
width=700
)
return fig
SensitivitySurfaceAnalyzer
# ABOUTME: Creates 2D sensitivity surfaces (contour/heatmaps) for two-variable analysis
# ABOUTME: Shows NPV as function of two parameters simultaneously
@dataclass
class SurfaceResult:
"""Results from 2D surface analysis."""
param1_name: str
param1_values: np.ndarray
param2_name: str
param2_values: np.ndarray
npv_surface: np.ndarray # 2D array of NPV values
base_npv: float
breakeven_contour: Optional[List[Tuple[float, float]]] = None
class SensitivitySurfaceAnalyzer:
"""
Generates 2D sensitivity surfaces showing NPV as function of two parameters.
Creates contour plots and heatmaps that reveal interaction effects
between parameters and identify breakeven combinations.
Usage:
analyzer = SensitivitySurfaceAnalyzer(npv_calculator)
result = analyzer.analyze(param1, param2, resolution=20)
fig = analyzer.create_contour_plot(result)
"""
def __init__(self, npv_calculator: Callable[[Dict[str, float]], float]):
"""
Initialize analyzer with NPV calculation function.
Args:
npv_calculator: Function that takes parameter dict and returns NPV
"""
self.npv_calculator = npv_calculator
self.base_params: Dict[str, float] = {}
def set_base_parameters(self, params: Dict[str, float]):
"""Set base case parameters for analysis."""
self.base_params = params.copy()
def analyze(
self,
param1: SensitivityParameter,
param2: SensitivityParameter,
resolution: int = 20,
find_breakeven: bool = True
) -> SurfaceResult:
"""
Generate 2D sensitivity surface.
Args:
param1: First parameter (x-axis)
param2: Second parameter (y-axis)
resolution: Number of points per axis
find_breakeven: Whether to find NPV=0 contour
Returns:
SurfaceResult with NPV surface data
"""
# Generate parameter grids
p1_values = np.array(param1.get_values())
p2_values = np.array(param2.get_values())
# If variations don't give enough points, interpolate
if len(p1_values) < resolution:
p1_values = np.linspace(p1_values.min(), p1_values.max(), resolution)
if len(p2_values) < resolution:
p2_values = np.linspace(p2_values.min(), p2_values.max(), resolution)
# Calculate NPV for each combination
npv_surface = np.zeros((len(p2_values), len(p1_values)))
for i, p2_val in enumerate(p2_values):
for j, p1_val in enumerate(p1_values):
test_params = self.base_params.copy()
test_params[param1.name] = p1_val
test_params[param2.name] = p2_val
npv_surface[i, j] = self.npv_calculator(test_params)
# Calculate base NPV
base_npv = self.npv_calculator(self.base_params)
# Find breakeven contour if requested
breakeven_contour = None
if find_breakeven:
breakeven_contour = self._find_breakeven_contour(
p1_values, p2_values, npv_surface
)
return SurfaceResult(
param1_name=param1.name,
param1_values=p1_values,
param2_name=param2.name,
param2_values=p2_values,
npv_surface=npv_surface,
base_npv=base_npv,
breakeven_contour=breakeven_contour
)
def _find_breakeven_contour(
self,
x_values: np.ndarray,
y_values: np.ndarray,
z_surface: np.ndarray
) -> List[Tuple[float, float]]:
"""Find points where NPV = 0 using linear interpolation."""
from scipy import ndimage
# Find zero-crossing contour
contour_points = []
# Check horizontal edges
for i in range(len(y_values)):
for j in range(len(x_values) - 1):
z1, z2 = z_surface[i, j], z_surface[i, j + 1]
if z1 * z2 < 0: # Sign change
# Linear interpolation
t = z1 / (z1 - z2)
x_cross = x_values[j] + t * (x_values[j + 1] - x_values[j])
contour_points.append((x_cross, y_values[i]))
# Check vertical edges
for i in range(len(y_values) - 1):
for j in range(len(x_values)):
z1, z2 = z_surface[i, j], z_surface[i + 1, j]
if z1 * z2 < 0: # Sign change
t = z1 / (z1 - z2)
y_cross = y_values[i] + t * (y_values[i + 1] - y_values[i])
contour_points.append((x_values[j], y_cross))
return contour_points
def create_contour_plot(
self,
result: SurfaceResult,
title: str = "NPV Sensitivity Surface",
show_breakeven: bool = True,
colorscale: str = "RdYlGn"
) -> go.Figure:
"""
Create interactive contour plot.
Args:
result: SurfaceResult from analyze()
title: Chart title
show_breakeven: Highlight NPV=0 contour
colorscale: Plotly colorscale name
Returns:
Plotly Figure object
"""
fig = go.Figure()
# Add contour surface
fig.add_trace(go.Contour(
z=result.npv_surface,
x=result.param1_values,
y=result.param2_values,
colorscale=colorscale,
contours=dict(
showlabels=True,
labelfont=dict(size=10, color='white')
),
colorbar=dict(title="NPV (MM$)"),
hovertemplate=(
f"{result.param1_name}: %{{x:.2f}}<br>"
f"{result.param2_name}: %{{y:.2f}}<br>"
"NPV: %{z:.2f} MM$<extra></extra>"
)
))
# Add breakeven contour if available
if show_breakeven and result.breakeven_contour:
x_be = [p[0] for p in result.breakeven_contour]
y_be = [p[1] for p in result.breakeven_contour]
fig.add_trace(go.Scatter(
x=x_be,
y=y_be,
mode='markers',
marker=dict(color='black', size=4, symbol='x'),
name='Breakeven (NPV=0)',
hovertemplate=(
f"{result.param1_name}: %{{x:.2f}}<br>"
f"{result.param2_name}: %{{y:.2f}}<br>"
"NPV: ~0<extra></extra>"
)
))
# Add base case marker
if result.param1_name in self.base_params and result.param2_name in self.base_params:
fig.add_trace(go.Scatter(
x=[self.base_params[result.param1_name]],
y=[self.base_params[result.param2_name]],
mode='markers',
marker=dict(color='blue', size=15, symbol='star'),
name=f'Base Case (NPV={result.base_npv:.1f})',
))
fig.update_layout(
title=dict(text=title, x=0.5),
xaxis_title=result.param1_name.replace("_", " ").title(),
yaxis_title=result.param2_name.replace("_", " ").title(),
height=600,
width=800
)
return fig
def create_heatmap(
self,
result: SurfaceResult,
title: str = "NPV Sensitivity Heatmap",
colorscale: str = "RdYlGn"
) -> go.Figure:
"""Create heatmap version of sensitivity surface."""
fig = go.Figure()
fig.add_trace(go.Heatmap(
z=result.npv_surface,
x=result.param1_values,
y=result.param2_values,
colorscale=colorscale,
colorbar=dict(title="NPV (MM$)"),
hovertemplate=(
f"{result.param1_name}: %{{x:.2f}}<br>"
f"{result.param2_name}: %{{y:.2f}}<br>"
"NPV: %{z:.2f} MM$<extra></extra>"
)
))
fig.update_layout(
title=dict(text=title, x=0.5),
xaxis_title=result.param1_name.replace("_", " ").title(),
yaxis_title=result.param2_name.replace("_", " ").title(),
height=600,
width=800
)
return fig
BreakevenAnalyzer
# ABOUTME: Calculates and visualizes breakeven prices for oil, gas, and costs
# ABOUTME: Supports multi-commodity breakeven analysis and sensitivity to other parameters
@dataclass
class BreakevenResult:
"""Results from breakeven analysis."""
parameter_name: str
breakeven_value: Optional[float]
unit: str
base_value: float
base_npv: float
sensitivity_curve: List[Tuple[float, float]] # [(param_value, npv), ...]
margin_at_base: float # % above/below breakeven
class BreakevenAnalyzer:
"""
Calculates breakeven prices and costs for investment decisions.
Finds the parameter value where NPV equals zero, indicating the
minimum/maximum acceptable value for project viability.
Usage:
analyzer = BreakevenAnalyzer(npv_calculator)
result = analyzer.find_breakeven("oil_price", base_value=70, range_pct=0.5)
fig = analyzer.create_breakeven_chart([oil_result, gas_result])
"""
def __init__(self, npv_calculator: Callable[[Dict[str, float]], float]):
"""Initialize with NPV calculator function."""
self.npv_calculator = npv_calculator
self.base_params: Dict[str, float] = {}
def set_base_parameters(self, params: Dict[str, float]):
"""Set base case parameters."""
self.base_params = params.copy()
def find_breakeven(
self,
param_name: str,
base_value: float,
range_pct: float = 0.5,
resolution: int = 50,
unit: str = ""
) -> BreakevenResult:
"""
Find breakeven value for a parameter using bisection.
Args:
param_name: Parameter to find breakeven for
base_value: Base case value
range_pct: Search range as fraction of base value
resolution: Points for sensitivity curve
unit: Display unit
Returns:
BreakevenResult with breakeven value and sensitivity data
"""
# Calculate base NPV
base_npv = self.npv_calculator(self.base_params)
# Generate sensitivity curve
min_val = base_value * (1 - range_pct)
max_val = base_value * (1 + range_pct)
test_values = np.linspace(min_val, max_val, resolution)
sensitivity_curve = []
for val in test_values:
test_params = self.base_params.copy()
test_params[param_name] = val
npv = self.npv_calculator(test_params)
sensitivity_curve.append((val, npv))
# Find breakeven using bisection
breakeven_value = self._bisection_search(
param_name, min_val, max_val, tolerance=0.01
)
# Calculate margin
if breakeven_value is not None:
margin = ((base_value - breakeven_value) / breakeven_value) * 100
else:
margin = float('inf') if base_npv > 0 else float('-inf')
return BreakevenResult(
parameter_name=param_name,
breakeven_value=breakeven_value,
unit=unit,
base_value=base_value,
base_npv=base_npv,
sensitivity_curve=sensitivity_curve,
margin_at_base=margin
)
def _bisection_search(
self,
param_name: str,
low: float,
high: float,
tolerance: float = 0.01,
max_iterations: int = 50
) -> Optional[float]:
"""Find breakeven using bisection method."""
# Check if breakeven exists in range
test_low = self.base_params.copy()
test_low[param_name] = low
npv_low = self.npv_calculator(test_low)
test_high = self.base_params.copy()
test_high[param_name] = high
npv_high = self.npv_calculator(test_high)
if npv_low * npv_high > 0:
# No sign change, breakeven not in range
return None
for _ in range(max_iterations):
mid = (low + high) / 2
test_mid = self.base_params.copy()
test_mid[param_name] = mid
npv_mid = self.npv_calculator(test_mid)
if abs(npv_mid) < tolerance or (high - low) / 2 < tolerance:
return mid
if npv_mid * npv_low < 0:
high = mid
npv_high = npv_mid
else:
low = mid
npv_low = npv_mid
return (low + high) / 2
def create_breakeven_chart(
self,
results: List[BreakevenResult],
title: str = "Breakeven Analysis"
) -> go.Figure:
"""
Create multi-parameter breakeven chart.
Args:
results: List of BreakevenResult objects
title: Chart title
Returns:
Plotly Figure with breakeven visualization
"""
fig = make_subplots(
rows=len(results), cols=1,
subplot_titles=[r.parameter_name.replace("_", " ").title() for r in results],
vertical_spacing=0.12
)
colors = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728']
for i, result in enumerate(results):
row = i + 1
x_vals = [p[0] for p in result.sensitivity_curve]
y_vals = [p[1] for p in result.sensitivity_curve]
# NPV curve
fig.add_trace(
go.Scatter(
x=x_vals, y=y_vals,
mode='lines',
line=dict(color=colors[i % len(colors)], width=2),
name=f'{result.parameter_name} Sensitivity',
showlegend=True
),
row=row, col=1
)
# Zero line
fig.add_hline(y=0, line_dash="dash", line_color="gray", row=row, col=1)
# Breakeven marker
if result.breakeven_value is not None:
fig.add_trace(
go.Scatter(
x=[result.breakeven_value], y=[0],
mode='markers+text',
marker=dict(color='red', size=12, symbol='x'),
text=[f'BE: {result.breakeven_value:.2f}{result.unit}'],
textposition='top center',
name=f'Breakeven',
showlegend=(i == 0)
),
row=row, col=1
)
# Base case marker
fig.add_trace(
go.Scatter(
x=[result.base_value], y=[result.base_npv],
mode='markers',
marker=dict(color='green', size=12, symbol='star'),
name=f'Base Case',
showlegend=(i == 0)
),
row=row, col=1
)
# Update axes
fig.update_xaxes(title_text=f"{result.parameter_name} ({result.unit})", row=row, col=1)
fig.update_yaxes(title_text="NPV (MM$)", row=row, col=1)
fig.update_layout(
title=dict(text=title, x=0.5),
height=300 * len(results),
showlegend=True
)
return fig
ScenarioMatrixAnalyzer
# ABOUTME: Creates scenario comparison matrices for management presentations
# ABOUTME: Supports low/mid/high scenarios with customizable probability weighting
@dataclass
class ScenarioDefinition:
"""Definition of a single scenario."""
name: str
parameters: Dict[str, float]
probability: float = 0.0 # For expected value calculation
color: str = "#1f77b4"
@dataclass
class ScenarioMatrixResult:
"""Results from scenario matrix analysis."""
scenarios: List[ScenarioDefinition]
npv_values: Dict[str, float] # scenario_name -> NPV
irr_values: Dict[str, float] # scenario_name -> IRR
payback_values: Dict[str, float] # scenario_name -> payback
expected_npv: float # Probability-weighted NPV
best_scenario: str
worst_scenario: str
class ScenarioMatrixAnalyzer:
"""
Creates scenario comparison matrices for investment decisions.
Compares multiple predefined scenarios (e.g., Low/Mid/High cases)
and calculates probability-weighted expected values.
Usage:
analyzer = ScenarioMatrixAnalyzer(npv_calculator, irr_calculator)
scenarios = [low_case, mid_case, high_case]
result = analyzer.analyze(scenarios)
fig = analyzer.create_comparison_chart(result)
"""
def __init__(
self,
npv_calculator: Callable[[Dict[str, float]], float],
irr_calculator: Optional[Callable[[Dict[str, float]], float]] = None,
payback_calculator: Optional[Callable[[Dict[str, float]], float]] = None
):
"""
Initialize with metric calculators.
Args:
npv_calculator: Function to calculate NPV
irr_calculator: Optional function to calculate IRR
payback_calculator: Optional function to calculate payback period
"""
self.npv_calculator = npv_calculator
self.irr_calculator = irr_calculator
self.payback_calculator = payback_calculator
def analyze(self, scenarios: List[ScenarioDefinition]) -> ScenarioMatrixResult:
"""
Analyze all scenarios and compare results.
Args:
scenarios: List of scenario definitions
Returns:
ScenarioMatrixResult with comparison data
"""
npv_values = {}
irr_values = {}
payback_values = {}
for scenario in scenarios:
# Calculate NPV
npv = self.npv_calculator(scenario.parameters)
npv_values[scenario.name] = npv
# Calculate IRR if available
if self.irr_calculator:
try:
irr = self.irr_calculator(scenario.parameters)
irr_values[scenario.name] = irr
except:
irr_values[scenario.name] = None
# Calculate payback if available
if self.payback_calculator:
try:
payback = self.payback_calculator(scenario.parameters)
payback_values[scenario.name] = payback
except:
payback_values[scenario.name] = None
# Calculate expected NPV
total_prob = sum(s.probability for s in scenarios)
if total_prob > 0:
expected_npv = sum(
s.probability * npv_values[s.name] / total_prob
for s in scenarios
)
else:
expected_npv = sum(npv_values.values()) / len(npv_values)
# Find best/worst scenarios
best_scenario = max(npv_values, key=npv_values.get)
worst_scenario = min(npv_values, key=npv_values.get)
return ScenarioMatrixResult(
scenarios=scenarios,
npv_values=npv_values,
irr_values=irr_values,
payback_values=payback_values,
expected_npv=expected_npv,
best_scenario=best_scenario,
worst_scenario=worst_scenario
)
def create_comparison_chart(
self,
result: ScenarioMatrixResult,
title: str = "Scenario Comparison"
) -> go.Figure:
"""
Create scenario comparison bar chart.
Args:
result: ScenarioMatrixResult from analyze()
title: Chart title
Returns:
Plotly Figure with scenario comparison
"""
scenarios = [s.name for s in result.scenarios]
npvs = [result.npv_values[s] for s in scenarios]
colors = [s.color for s in result.scenarios]
fig = go.Figure()
# NPV bars
fig.add_trace(go.Bar(
x=scenarios,
y=npvs,
marker_color=colors,
text=[f'${npv:.1f}M' for npv in npvs],
textposition='outside',
name='NPV'
))
# Expected value line
fig.add_hline(
y=result.expected_npv,
line_dash="dash",
line_color="black",
annotation_text=f"Expected: ${result.expected_npv:.1f}M",
annotation_position="right"
)
# Zero line
fig.add_hline(y=0, line_color="gray", line_width=1)
fig.update_layout(
title=dict(text=title, x=0.5),
xaxis_title="Scenario",
yaxis_title="NPV (MM$)",
height=500,
showlegend=False
)
return fig
def create_matrix_table(
self,
result: ScenarioMatrixResult
) -> go.Figure:
"""
Create scenario matrix as formatted table.
Returns:
Plotly Figure with table visualization
"""
scenarios = [s.name for s in result.scenarios]
# Build table data
headers = ['Metric'] + scenarios
npv_row = ['NPV (MM$)'] + [f'{result.npv_values[s]:.1f}' for s in scenarios]
rows = [npv_row]
if result.irr_values:
irr_row = ['IRR (%)'] + [
f'{result.irr_values[s]*100:.1f}' if result.irr_values.get(s) else 'N/A'
for s in scenarios
]
rows.append(irr_row)
if result.payback_values:
payback_row = ['Payback (yrs)'] + [
f'{result.payback_values[s]:.1f}' if result.payback_values.get(s) else 'N/A'
for s in scenarios
]
rows.append(payback_row)
prob_row = ['Probability'] + [f'{s.probability*100:.0f}%' for s in result.scenarios]
rows.append(prob_row)
# Create table
fig = go.Figure(data=[go.Table(
header=dict(
values=headers,
fill_color='#1f77b4',
font=dict(color='white', size=12),
align='center'
),
cells=dict(
values=[[row[i] for row in rows] for i in range(len(headers))],
fill_color=[['white', '#f0f0f0'] * (len(rows) // 2 + 1)][:len(rows)],
align='center'
)
)])
fig.update_layout(
title=dict(text="Scenario Matrix", x=0.5),
height=200 + 40 * len(rows)
)
return fig
DecisionTreeAnalyzer
# ABOUTME: Creates decision tree analysis for staged investments
# ABOUTME: Calculates expected value considering probabilities and decision points
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Union
import plotly.graph_objects as go
@dataclass
class DecisionNode:
"""A decision or chance node in the tree."""
name: str
node_type: str # "decision", "chance", "terminal"
value: Optional[float] = None # For terminal nodes
probability: float = 1.0 # For chance nodes
children: List['DecisionNode'] = field(default_factory=list)
investment: float = 0.0 # Investment required at this node
expected_value: float = 0.0 # Calculated expected value
class DecisionTreeAnalyzer:
"""
Performs decision tree analysis for staged investment decisions.
Useful for analyzing:
- Drill/don't drill decisions
- Appraisal well outcomes
- Development staging (Phase 1, Phase 2, etc.)
- Farmout vs. operate decisions
Usage:
analyzer = DecisionTreeAnalyzer()
tree = analyzer.build_drill_decision_tree(
drill_cost=10, success_prob=0.3,
success_npv=100, failure_npv=-10
)
result = analyzer.evaluate(tree)
fig = analyzer.visualize(tree)
"""
def build_drill_decision_tree(
self,
drill_cost: float,
success_prob: float,
success_npv: float,
failure_npv: float,
farmout_npv: float = 0
) -> DecisionNode:
"""
Build standard drill/farmout/don't drill decision tree.
Args:
drill_cost: Cost to drill (MM$)
success_prob: Probability of success (0-1)
success_npv: NPV if successful (MM$)
failure_npv: NPV if dry hole (MM$, usually negative)
farmout_npv: NPV if farmout (MM$)
Returns:
Root DecisionNode of tree
"""
# Success terminal node
success = DecisionNode(
name="Success",
node_type="terminal",
value=success_npv,
probability=success_prob
)
# Failure terminal node
failure = DecisionNode(
name="Dry Hole",
node_type="terminal",
value=failure_npv,
probability=1 - success_prob
)
# Drill chance node
drill_outcome = DecisionNode(
name="Drill Outcome",
node_type="chance",
children=[success, failure],
investment=drill_cost
)
# Farmout terminal
farmout = DecisionNode(
name="Farmout",
node_type="terminal",
value=farmout_npv
)
# Don't drill terminal
no_drill = DecisionNode(
name="Don't Drill",
node_type="terminal",
value=0
)
# Root decision node
root = DecisionNode(
name="Investment Decision",
node_type="decision",
children=[drill_outcome, farmout, no_drill]
)
return root
def evaluate(self, node: DecisionNode) -> float:
"""
Evaluate decision tree using backward induction.
Args:
node: Root node of tree
Returns:
Expected value at root node
"""
if node.node_type == "terminal":
node.expected_value = node.value or 0
return node.expected_value
# Recursively evaluate children
for child in node.children:
self.evaluate(child)
if node.node_type == "chance":
# Expected value = sum of prob * value
ev = sum(c.probability * c.expected_value for c in node.children)
node.expected_value = ev - node.investment
elif node.node_type == "decision":
# Optimal decision = max expected value
node.expected_value = max(c.expected_value for c in node.children)
return node.expected_value
def get_optimal_path(self, node: DecisionNode) -> List[str]:
"""Get the optimal decision path from root."""
path = [node.name]
if node.node_type == "terminal":
return path
if node.node_type == "decision":
# Find child with max expected value
best_child = max(node.children, key=lambda c: c.expected_value)
path.extend(self.get_optimal_path(best_child))
elif node.node_type == "chance":
# For chance nodes, show expected outcome
for child in node.children:
path.append(f" {child.name} (p={child.probability:.0%})")
return path
def visualize(
self,
root: DecisionNode,
title: str = "Decision Tree Analysis"
) -> go.Figure:
"""
Create decision tree visualization using Sankey diagram.
Args:
root: Root node of evaluated tree
title: Chart title
Returns:
Plotly Figure with decision tree
"""
# Collect all nodes and links
nodes = []
links = []
node_index = {}
def add_node(node: DecisionNode, depth: int = 0):
if node.name not in node_index:
idx = len(nodes)
node_index[node.name] = idx
# Color by node type
if node.node_type == "decision":
color = "#1f77b4" # Blue
elif node.node_type == "chance":
color = "#ff7f0e" # Orange
else: # terminal
color = "#2ca02c" if (node.value or 0) > 0 else "#d62728"
# Label with expected value
if node.node_type == "terminal":
label = f"{node.name}<br>${node.value:.1f}M"
else:
label = f"{node.name}<br>EV: ${node.expected_value:.1f}M"
nodes.append({
'label': label,
'color': color,
'depth': depth
})
return node_index[node.name]
def process_tree(node: DecisionNode, depth: int = 0):
source_idx = add_node(node, depth)
for child in node.children:
target_idx = add_node(child, depth + 1)
# Link value is the flow "magnitude"
value = abs(child.expected_value) if child.expected_value else 1
# Label for link
if child.probability < 1:
label = f"{child.probability:.0%}"
else:
label = ""
links.append({
'source': source_idx,
'target': target_idx,
'value': max(value, 0.1), # Minimum for visibility
'label': label
})
process_tree(child, depth + 1)
process_tree(root)
fig = go.Figure(data=[go.Sankey(
node=dict(
pad=15,
thickness=20,
line=dict(color="black", width=0.5),
label=[n['label'] for n in nodes],
color=[n['color'] for n in nodes]
),
link=dict(
source=[l['source'] for l in links],
target=[l['target'] for l in links],
value=[l['value'] for l in links],
label=[l['label'] for l in links]
)
)])
fig.update_layout(
title=dict(text=title, x=0.5),
height=500,
font_size=10
)
return fig
SensitivityDashboard
# ABOUTME: Combines all sensitivity analyses into comprehensive dashboard
# ABOUTME: Creates executive-ready reports with multiple visualization types
from typing import Dict, List, Optional, Callable
import plotly.graph_objects as go
from plotly.subplots import make_subplots
class SensitivityDashboard:
"""
Creates comprehensive sensitivity analysis dashboard.
Combines spider diagrams, contour plots, tornado charts, and
breakeven analysis into a single executive summary.
Usage:
dashboard = SensitivityDashboard(
npv_calculator=calc_npv,
…(truncated)