vamseeachanta NPV Analyzer
Perform comprehensive Net Present Value (NPV) analysis and economic evaluation for oil & gas development projects, including cash flow modeling, price scenarios, Monte Carlo simulation for probabilistic analysis, and financial metrics.
When to Use
- Calculating NPV for field development projects
- Modeling cash flows with production forecasts
- Running oil/gas price scenario analysis (low/mid/high)
- Monte Carlo simulation for P10/P50/P90 NPV estimates
- Probabilistic risk analysis with multiple input distributions
- Applying working interest and royalty calculations
- Evaluating different development types (subsea, platform, FPSO)
- Computing IRR, payback period, and profitability index
- Calculating Value at Risk (VaR) and Expected Shortfall
- Comparing economic outcomes across multiple scenarios
Core Pattern
Production Forecast → Price Assumptions → Cash Flow Model → Discount → Metrics
Implementation
Data Models
from dataclasses import dataclass, field
from datetime import date
from typing import Optional, List, Dict, Tuple
from enum import Enum
import numpy as np
import pandas as pd
class DevelopmentType(Enum):
"""Field development concepts."""
SUBSEA_TIEBACK = "subsea_tieback"
FIXED_PLATFORM = "fixed_platform"
FPSO = "fpso"
SPAR = "spar"
TLP = "tlp"
SEMI = "semi_submersible"
class PriceScenario(Enum):
"""Oil/gas price scenarios."""
LOW = "low"
MID = "mid"
HIGH = "high"
STRIP = "strip" # Forward curve prices
CUSTOM = "custom"
@dataclass
class PriceAssumptions:
"""Oil and gas price assumptions."""
scenario: PriceScenario = PriceScenario.MID
oil_price_usd_bbl: float = 70.0
gas_price_usd_mmbtu: float = 3.50
ngl_price_usd_bbl: float = 35.0
# Price escalation
annual_escalation_pct: float = 2.0
# Price deck for forward curve
price_deck: Optional[Dict[int, Tuple[float, float]]] = None # year: (oil, gas)
def get_prices(self, year: int, base_year: int = None) -> Tuple[float, float]:
"""Get escalated prices for a specific year."""
if self.price_deck and year in self.price_deck:
return self.price_deck[year]
base_year = base_year or date.today().year
years_forward = year - base_year
escalation = (1 + self.annual_escalation_pct / 100) ** years_forward
oil_price = self.oil_price_usd_bbl * escalation
gas_price = self.gas_price_usd_mmbtu * escalation
return oil_price, gas_price
@dataclass
class FiscalTerms:
"""Fiscal and ownership terms."""
working_interest: float = 1.0 # Fraction (0-1)
net_revenue_interest: float = 0.875 # After royalty
royalty_rate: float = 0.125 # 12.5% federal royalty
state_severance_tax: float = 0.0 # Varies by state
federal_income_tax: float = 0.21 # Corporate rate
ad_valorem_tax: float = 0.0
# Cost recovery (for PSC contracts)
cost_recovery_limit: float = 1.0 # 100% for typical US
profit_oil_share: float = 1.0 # 100% for typical US
def net_revenue_factor(self) -> float:
"""Calculate combined net revenue factor."""
return self.working_interest * self.net_revenue_interest
@dataclass
class CapexSchedule:
"""Capital expenditure schedule."""
drilling_per_well: float = 0.0 # USD millions
completion_per_well: float = 0.0
facilities: float = 0.0
subsea_equipment: float = 0.0
pipeline: float = 0.0
other_capex: float = 0.0
# Timing (years from sanction)
schedule: Dict[int, float] = field(default_factory=dict) # year: capex
@property
def total_capex(self) -> float:
"""Total capital expenditure."""
return (self.drilling_per_well + self.completion_per_well +
self.facilities + self.subsea_equipment +
self.pipeline + self.other_capex)
@dataclass
class OpexAssumptions:
"""Operating expenditure assumptions."""
fixed_opex_usd_year: float = 0.0 # Annual fixed costs
variable_opex_usd_boe: float = 0.0 # Per BOE
workover_budget: float = 0.0 # Annual workover allowance
insurance: float = 0.0
g_and_a: float = 0.0 # General & administrative
# Abandonment
abandonment_cost: float = 0.0
abandonment_year: Optional[int] = None
@dataclass
class ProductionForecast:
"""Production forecast by year."""
year: int
oil_mbbls: float = 0.0 # Thousand barrels
gas_mmcf: float = 0.0 # Million cubic feet
ngl_mbbls: float = 0.0
water_mbbls: float = 0.0
@property
def boe(self) -> float:
"""Barrels of oil equivalent (6:1 gas)."""
return self.oil_mbbls + self.ngl_mbbls + (self.gas_mmcf / 6.0)
@property
def gas_mmbtu(self) -> float:
"""Gas in MMBTU (assuming 1 MCF = 1 MMBTU)."""
return self.gas_mmcf * 1000 # Convert MMCF to MMBTU
@dataclass
class CashFlowYear:
"""Annual cash flow calculation."""
year: int
# Revenue
gross_oil_revenue: float = 0.0
gross_gas_revenue: float = 0.0
gross_ngl_revenue: float = 0.0
total_gross_revenue: float = 0.0
net_revenue: float = 0.0
# Costs
capex: float = 0.0
opex: float = 0.0
severance_tax: float = 0.0
ad_valorem_tax: float = 0.0
abandonment: float = 0.0
# Taxes
taxable_income: float = 0.0
income_tax: float = 0.0
# Cash flows
before_tax_cash_flow: float = 0.0
after_tax_cash_flow: float = 0.0
cumulative_cash_flow: float = 0.0
discounted_cash_flow: float = 0.0
NPV Calculator
from typing import List, Dict, Optional
import numpy as np
import numpy_financial as npf
class NPVCalculator:
"""
Calculate NPV and related economic metrics for oil & gas projects.
"""
def __init__(self,
production: List[ProductionForecast],
prices: PriceAssumptions,
fiscal: FiscalTerms,
capex: CapexSchedule,
opex: OpexAssumptions,
discount_rate: float = 0.10):
"""
Initialize NPV calculator.
Args:
production: List of annual production forecasts
prices: Price assumptions
fiscal: Fiscal and ownership terms
capex: Capital expenditure schedule
opex: Operating expenditure assumptions
discount_rate: Annual discount rate (default 10%)
"""
self.production = sorted(production, key=lambda x: x.year)
self.prices = prices
self.fiscal = fiscal
self.capex = capex
self.opex = opex
self.discount_rate = discount_rate
self._cash_flows: Optional[List[CashFlowYear]] = None
self._base_year: Optional[int] = None
@property
def base_year(self) -> int:
"""Get or set base year for discounting."""
if self._base_year is None:
self._base_year = min(p.year for p in self.production)
return self._base_year
@base_year.setter
def base_year(self, value: int):
self._base_year = value
self._cash_flows = None # Reset calculations
def calculate_cash_flows(self) -> List[CashFlowYear]:
"""Calculate annual cash flows."""
if self._cash_flows is not None:
return self._cash_flows
cash_flows = []
cumulative = 0.0
for prod in self.production:
cf = CashFlowYear(year=prod.year)
# Get prices for year
oil_price, gas_price = self.prices.get_prices(prod.year, self.base_year)
ngl_price = self.prices.ngl_price_usd_bbl * (
(1 + self.prices.annual_escalation_pct / 100) **
(prod.year - self.base_year)
)
# Gross revenue (in thousands USD, since production in Mbbls/MMCF)
cf.gross_oil_revenue = prod.oil_mbbls * oil_price * 1000
cf.gross_gas_revenue = prod.gas_mmbtu * gas_price / 1000 # MMBTU to $ thousands
cf.gross_ngl_revenue = prod.ngl_mbbls * ngl_price * 1000
cf.total_gross_revenue = (cf.gross_oil_revenue +
cf.gross_gas_revenue +
cf.gross_ngl_revenue)
# Net revenue (apply WI and NRI)
cf.net_revenue = cf.total_gross_revenue * self.fiscal.net_revenue_factor()
# Capital expenditure
cf.capex = self.capex.schedule.get(prod.year, 0.0) * 1e6 # Convert to $
# Operating expenditure
cf.opex = (
self.opex.fixed_opex_usd_year +
self.opex.variable_opex_usd_boe * prod.boe * 1000 +
self.opex.workover_budget +
self.opex.insurance +
self.opex.g_and_a
)
# Abandonment
if (self.opex.abandonment_year and
prod.year == self.opex.abandonment_year):
cf.abandonment = self.opex.abandonment_cost
# Severance tax (on gross revenue before NRI)
cf.severance_tax = (cf.total_gross_revenue *
self.fiscal.working_interest *
self.fiscal.state_severance_tax)
# Ad valorem tax
cf.ad_valorem_tax = (cf.total_gross_revenue *
self.fiscal.working_interest *
self.fiscal.ad_valorem_tax)
# Before tax cash flow
cf.before_tax_cash_flow = (cf.net_revenue -
cf.capex -
cf.opex -
cf.severance_tax -
cf.ad_valorem_tax -
cf.abandonment)
# Taxable income (simplified - no DD&A or intangible drilling)
cf.taxable_income = max(0, cf.before_tax_cash_flow)
cf.income_tax = cf.taxable_income * self.fiscal.federal_income_tax
# After tax cash flow
cf.after_tax_cash_flow = cf.before_tax_cash_flow - cf.income_tax
# Cumulative cash flow
cumulative += cf.after_tax_cash_flow
cf.cumulative_cash_flow = cumulative
# Discounted cash flow
years_from_base = prod.year - self.base_year
discount_factor = 1 / (1 + self.discount_rate) ** years_from_base
cf.discounted_cash_flow = cf.after_tax_cash_flow * discount_factor
cash_flows.append(cf)
self._cash_flows = cash_flows
return cash_flows
def npv(self) -> float:
"""Calculate Net Present Value."""
cash_flows = self.calculate_cash_flows()
return sum(cf.discounted_cash_flow for cf in cash_flows)
def irr(self) -> Optional[float]:
"""Calculate Internal Rate of Return."""
cash_flows = self.calculate_cash_flows()
cf_values = [cf.after_tax_cash_flow for cf in cash_flows]
try:
return npf.irr(cf_values)
except:
return None
def payback_period(self) -> Optional[float]:
"""Calculate payback period in years."""
cash_flows = self.calculate_cash_flows()
for i, cf in enumerate(cash_flows):
if cf.cumulative_cash_flow >= 0:
if i == 0:
return 0.0
# Interpolate
prev_cf = cash_flows[i - 1]
fraction = (-prev_cf.cumulative_cash_flow /
(cf.cumulative_cash_flow - prev_cf.cumulative_cash_flow))
return (cf.year - cash_flows[0].year - 1) + fraction
return None # Never pays back
def discounted_payback(self) -> Optional[float]:
"""Calculate discounted payback period."""
cash_flows = self.calculate_cash_flows()
cumulative_dcf = 0.0
for i, cf in enumerate(cash_flows):
cumulative_dcf += cf.discounted_cash_flow
if cumulative_dcf >= 0:
if i == 0:
return 0.0
prev_cumulative = cumulative_dcf - cf.discounted_cash_flow
fraction = (-prev_cumulative / cf.discounted_cash_flow)
return (cf.year - cash_flows[0].year - 1) + fraction
return None
def profitability_index(self) -> float:
"""Calculate profitability index (PI = NPV / Investment + 1)."""
cash_flows = self.calculate_cash_flows()
total_investment = sum(cf.capex for cf in cash_flows)
if total_investment == 0:
return float('inf')
return (self.npv() / total_investment) + 1
def investment_efficiency(self) -> float:
"""Calculate investment efficiency (NPV per $ invested)."""
cash_flows = self.calculate_cash_flows()
total_investment = sum(cf.capex for cf in cash_flows)
if total_investment == 0:
return float('inf')
return self.npv() / total_investment
def breakeven_price(self, commodity: str = 'oil',
tolerance: float = 0.01) -> Optional[float]:
"""
Calculate breakeven price for NPV = 0.
Args:
commodity: 'oil' or 'gas'
tolerance: NPV tolerance for convergence
"""
from scipy.optimize import brentq
original_prices = self.prices
def npv_at_price(price):
if commodity == 'oil':
self.prices = PriceAssumptions(
oil_price_usd_bbl=price,
gas_price_usd_mmbtu=original_prices.gas_price_usd_mmbtu
)
else:
self.prices = PriceAssumptions(
oil_price_usd_bbl=original_prices.oil_price_usd_bbl,
gas_price_usd_mmbtu=price
)
self._cash_flows = None
return self.npv()
try:
# Search between $10 and $200 for oil, $1 and $20 for gas
if commodity == 'oil':
breakeven = brentq(npv_at_price, 10, 200)
else:
breakeven = brentq(npv_at_price, 1, 20)
except:
breakeven = None
finally:
self.prices = original_prices
self._cash_flows = None
return breakeven
def to_dataframe(self) -> pd.DataFrame:
"""Export cash flows to DataFrame."""
cash_flows = self.calculate_cash_flows()
data = []
for cf in cash_flows:
data.append({
'year': cf.year,
'gross_revenue': cf.total_gross_revenue,
'net_revenue': cf.net_revenue,
'capex': cf.capex,
'opex': cf.opex,
'taxes': cf.severance_tax + cf.ad_valorem_tax + cf.income_tax,
'btcf': cf.before_tax_cash_flow,
'atcf': cf.after_tax_cash_flow,
'cumulative': cf.cumulative_cash_flow,
'dcf': cf.discounted_cash_flow
})
return pd.DataFrame(data)
def summary(self) -> Dict:
"""Generate summary of economic metrics."""
cash_flows = self.calculate_cash_flows()
return {
'npv_mm': self.npv() / 1e6,
'irr_pct': (self.irr() or 0) * 100,
'payback_years': self.payback_period(),
'discounted_payback_years': self.discounted_payback(),
'profitability_index': self.profitability_index(),
'investment_efficiency': self.investment_efficiency(),
'total_capex_mm': sum(cf.capex for cf in cash_flows) / 1e6,
'total_opex_mm': sum(cf.opex for cf in cash_flows) / 1e6,
'total_revenue_mm': sum(cf.net_revenue for cf in cash_flows) / 1e6,
'cumulative_production_mmboe': sum(p.boe for p in self.production) / 1000,
'discount_rate_pct': self.discount_rate * 100
}
Scenario Analyzer
from typing import List, Dict
from dataclasses import replace
import pandas as pd
class ScenarioAnalyzer:
"""
Run multiple NPV scenarios with different price assumptions.
"""
# Standard price scenarios
PRICE_SCENARIOS = {
PriceScenario.LOW: PriceAssumptions(
scenario=PriceScenario.LOW,
oil_price_usd_bbl=50.0,
gas_price_usd_mmbtu=2.50
),
PriceScenario.MID: PriceAssumptions(
scenario=PriceScenario.MID,
oil_price_usd_bbl=70.0,
gas_price_usd_mmbtu=3.50
),
PriceScenario.HIGH: PriceAssumptions(
scenario=PriceScenario.HIGH,
oil_price_usd_bbl=90.0,
gas_price_usd_mmbtu=5.00
)
}
def __init__(self,
production: List[ProductionForecast],
fiscal: FiscalTerms,
capex: CapexSchedule,
opex: OpexAssumptions,
discount_rate: float = 0.10):
"""Initialize scenario analyzer."""
self.production = production
self.fiscal = fiscal
self.capex = capex
self.opex = opex
self.discount_rate = discount_rate
def run_price_scenarios(self) -> pd.DataFrame:
"""Run standard price scenarios and compare results."""
results = []
for scenario, prices in self.PRICE_SCENARIOS.items():
calc = NPVCalculator(
production=self.production,
prices=prices,
fiscal=self.fiscal,
capex=self.capex,
opex=self.opex,
discount_rate=self.discount_rate
)
summary = calc.summary()
summary['scenario'] = scenario.value
summary['oil_price'] = prices.oil_price_usd_bbl
summary['gas_price'] = prices.gas_price_usd_mmbtu
results.append(summary)
return pd.DataFrame(results)
def sensitivity_analysis(self,
base_prices: PriceAssumptions,
variable: str,
range_pct: float = 0.30,
steps: int = 11) -> pd.DataFrame:
"""
Run sensitivity analysis on a single variable.
Args:
base_prices: Base case prices
variable: Variable to vary ('oil_price', 'gas_price', 'capex', 'opex', 'discount_rate')
range_pct: Range as percentage of base value
steps: Number of steps
"""
results = []
# Get base value
if variable == 'oil_price':
base_value = base_prices.oil_price_usd_bbl
elif variable == 'gas_price':
base_value = base_prices.gas_price_usd_mmbtu
elif variable == 'capex':
base_value = 1.0 # Multiplier
elif variable == 'opex':
base_value = 1.0 # Multiplier
elif variable == 'discount_rate':
base_value = self.discount_rate
else:
raise ValueError(f"Unknown variable: {variable}")
# Generate range
min_val = base_value * (1 - range_pct)
max_val = base_value * (1 + range_pct)
values = np.linspace(min_val, max_val, steps)
for val in values:
# Modify parameter
if variable == 'oil_price':
prices = replace(base_prices, oil_price_usd_bbl=val)
capex = self.capex
opex = self.opex
discount = self.discount_rate
elif variable == 'gas_price':
prices = replace(base_prices, gas_price_usd_mmbtu=val)
capex = self.capex
opex = self.opex
discount = self.discount_rate
elif variable == 'capex':
prices = base_prices
# Scale capex schedule
capex = replace(self.capex,
schedule={y: c * val for y, c in self.capex.schedule.items()})
opex = self.opex
discount = self.discount_rate
elif variable == 'opex':
prices = base_prices
capex = self.capex
opex = replace(self.opex,
fixed_opex_usd_year=self.opex.fixed_opex_usd_year * val,
variable_opex_usd_boe=self.opex.variable_opex_usd_boe * val)
discount = self.discount_rate
elif variable == 'discount_rate':
prices = base_prices
capex = self.capex
opex = self.opex
discount = val
# Calculate NPV
calc = NPVCalculator(
production=self.production,
prices=prices,
fiscal=self.fiscal,
capex=capex,
opex=opex,
discount_rate=discount
)
results.append({
'variable': variable,
'value': val,
'pct_change': (val - base_value) / base_value * 100,
'npv_mm': calc.npv() / 1e6,
'irr_pct': (calc.irr() or 0) * 100
})
return pd.DataFrame(results)
def tornado_chart_data(self, base_prices: PriceAssumptions,
range_pct: float = 0.20) -> pd.DataFrame:
"""Generate data for tornado sensitivity chart."""
variables = ['oil_price', 'gas_price', 'capex', 'opex', 'discount_rate']
results = []
# Calculate base case
base_calc = NPVCalculator(
production=self.production,
prices=base_prices,
fiscal=self.fiscal,
capex=self.capex,
opex=self.opex,
discount_rate=self.discount_rate
)
base_npv = base_calc.npv() / 1e6
for var in variables:
sensitivity = self.sensitivity_analysis(base_prices, var, range_pct, steps=3)
low = sensitivity.iloc[0]['npv_mm']
high = sensitivity.iloc[-1]['npv_mm']
results.append({
'variable': var,
'low_npv': low,
'high_npv': high,
'base_npv': base_npv,
'range': abs(high - low)
})
return pd.DataFrame(results).sort_values('range', ascending=False)
Monte Carlo Simulator
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Tuple
from enum import Enum
import numpy as np
import pandas as pd
class DistributionType(Enum):
"""Probability distribution types for Monte Carlo inputs."""
NORMAL = "normal"
LOGNORMAL = "lognormal"
TRIANGULAR = "triangular"
UNIFORM = "uniform"
PERT = "pert"
@dataclass
class DistributionParams:
"""Parameters for probability distributions."""
distribution: DistributionType = DistributionType.TRIANGULAR
min_value: float = 0.0
mode_value: float = 0.0 # Most likely (for triangular/PERT)
max_value: float = 0.0
mean: Optional[float] = None # For normal/lognormal
std_dev: Optional[float] = None
def sample(self, n_samples: int, rng: np.random.Generator = None) -> np.ndarray:
"""Generate random samples from the distribution."""
if rng is None:
rng = np.random.default_rng()
if self.distribution == DistributionType.UNIFORM:
return rng.uniform(self.min_value, self.max_value, n_samples)
elif self.distribution == DistributionType.TRIANGULAR:
return rng.triangular(self.min_value, self.mode_value, self.max_value, n_samples)
elif self.distribution == DistributionType.NORMAL:
mean = self.mean or self.mode_value
std = self.std_dev or (self.max_value - self.min_value) / 6
samples = rng.normal(mean, std, n_samples)
return np.clip(samples, self.min_value, self.max_value)
elif self.distribution == DistributionType.LOGNORMAL:
mean = self.mean or self.mode_value
std = self.std_dev or 0.2
samples = rng.lognormal(np.log(mean), std, n_samples)
return np.clip(samples, self.min_value, self.max_value)
elif self.distribution == DistributionType.PERT:
# PERT uses Beta distribution with shape parameters
alpha = 1 + 4 * (self.mode_value - self.min_value) / (self.max_value - self.min_value)
beta = 1 + 4 * (self.max_value - self.mode_value) / (self.max_value - self.min_value)
samples = rng.beta(alpha, beta, n_samples)
return self.min_value + samples * (self.max_value - self.min_value)
raise ValueError(f"Unknown distribution: {self.distribution}")
@dataclass
class MonteCarloConfig:
"""Configuration for Monte Carlo simulation."""
n_iterations: int = 1000
random_seed: Optional[int] = None
# Price distributions
oil_price: Optional[DistributionParams] = None
gas_price: Optional[DistributionParams] = None
# Cost multiplier distributions (1.0 = base case)
capex_multiplier: Optional[DistributionParams] = None
opex_multiplier: Optional[DistributionParams] = None
# Production multiplier distribution (1.0 = base case)
production_multiplier: Optional[DistributionParams] = None
# Discount rate distribution
discount_rate: Optional[DistributionParams] = None
# Correlation matrix (optional, for correlated variables)
correlations: Optional[Dict[str, Dict[str, float]]] = None
@dataclass
class MonteCarloResult:
"""Results from Monte Carlo simulation."""
npv_values: np.ndarray
irr_values: np.ndarray
payback_values: np.ndarray
# Percentiles
p10_npv: float = 0.0 # 10th percentile (pessimistic)
p50_npv: float = 0.0 # 50th percentile (median)
p90_npv: float = 0.0 # 90th percentile (optimistic)
p10_irr: float = 0.0
p50_irr: float = 0.0
p90_irr: float = 0.0
# Statistics
mean_npv: float = 0.0
std_npv: float = 0.0
prob_positive_npv: float = 0.0
# Input samples (for correlation analysis)
input_samples: Optional[pd.DataFrame] = None
def summary(self) -> Dict:
"""Generate summary statistics."""
return {
'mean_npv_mm': self.mean_npv / 1e6,
'std_npv_mm': self.std_npv / 1e6,
'p10_npv_mm': self.p10_npv / 1e6,
'p50_npv_mm': self.p50_npv / 1e6,
'p90_npv_mm': self.p90_npv / 1e6,
'p10_irr_pct': self.p10_irr * 100,
'p50_irr_pct': self.p50_irr * 100,
'p90_irr_pct': self.p90_irr * 100,
'prob_positive_npv_pct': self.prob_positive_npv * 100,
'n_iterations': len(self.npv_values)
}
class MonteCarloSimulator:
"""
Run Monte Carlo simulations for probabilistic NPV analysis.
Supports:
- Multiple probability distributions (normal, triangular, PERT, uniform, lognormal)
- Correlated input variables
- P10/P50/P90 output generation
- Risk analysis and probability of success
"""
def __init__(self,
production: List[ProductionForecast],
base_prices: PriceAssumptions,
fiscal: FiscalTerms,
capex: CapexSchedule,
opex: OpexAssumptions,
base_discount_rate: float = 0.10):
"""
Initialize Monte Carlo simulator.
Args:
production: Base case production forecast
base_prices: Base case price assumptions
fiscal: Fiscal terms (fixed)
capex: Base case capital expenditure
opex: Base case operating expenditure
base_discount_rate: Base case discount rate
"""
self.production = production
self.base_prices = base_prices
self.fiscal = fiscal
self.capex = capex
self.opex = opex
self.base_discount_rate = base_discount_rate
def run_simulation(self, config: MonteCarloConfig) -> MonteCarloResult:
"""
Run Monte Carlo simulation.
Args:
config: Simulation configuration with distributions
Returns:
MonteCarloResult with NPV distribution and statistics
"""
rng = np.random.default_rng(config.random_seed)
n = config.n_iterations
# Generate input samples
samples = self._generate_samples(config, n, rng)
# Run iterations
npv_values = np.zeros(n)
irr_values = np.zeros(n)
payback_values = np.zeros(n)
for i in range(n):
# Build modified inputs for this iteration
prices = self._modify_prices(samples, i)
modified_capex = self._modify_capex(samples, i)
modified_opex = self._modify_opex(samples, i)
modified_production = self._modify_production(samples, i)
discount = samples.get('discount_rate', [self.base_discount_rate] * n)[i]
# Calculate NPV for this iteration
calc = NPVCalculator(
production=modified_production,
prices=prices,
fiscal=self.fiscal,
capex=modified_capex,
opex=modified_opex,
discount_rate=discount
)
npv_values[i] = calc.npv()
irr_values[i] = calc.irr() or 0.0
payback_values[i] = calc.payback_period() or float('inf')
# Calculate statistics
result = MonteCarloResult(
npv_values=npv_values,
irr_values=irr_values,
payback_values=payback_values,
p10_npv=np.percentile(npv_values, 10),
p50_npv=np.percentile(npv_values, 50),
p90_npv=np.percentile(npv_values, 90),
p10_irr=np.percentile(irr_values, 10),
p50_irr=np.percentile(irr_values, 50),
p90_irr=np.percentile(irr_values, 90),
mean_npv=np.mean(npv_values),
std_npv=np.std(npv_values),
prob_positive_npv=np.sum(npv_values > 0) / n,
input_samples=pd.DataFrame(samples)
)
return result
def _generate_samples(self, config: MonteCarloConfig, n: int,
rng: np.random.Generator) -> Dict[str, np.ndarray]:
"""Generate random samples for all input variables."""
samples = {}
if config.oil_price:
samples['oil_price'] = config.oil_price.sample(n, rng)
else:
samples['oil_price'] = np.full(n, self.base_prices.oil_price_usd_bbl)
if config.gas_price:
samples['gas_price'] = config.gas_price.sample(n, rng)
else:
samples['gas_price'] = np.full(n, self.base_prices.gas_price_usd_mmbtu)
if config.capex_multiplier:
samples['capex_mult'] = config.capex_multiplier.sample(n, rng)
else:
samples['capex_mult'] = np.ones(n)
if config.opex_multiplier:
samples['opex_mult'] = config.opex_multiplier.sample(n, rng)
else:
samples['opex_mult'] = np.ones(n)
if config.production_multiplier:
samples['prod_mult'] = config.production_multiplier.sample(n, rng)
else:
samples['prod_mult'] = np.ones(n)
if config.discount_rate:
samples['discount_rate'] = config.discount_rate.sample(n, rng)
else:
samples['discount_rate'] = np.full(n, self.base_discount_rate)
return samples
def _modify_prices(self, samples: Dict, index: int) -> PriceAssumptions:
"""Create modified price assumptions for iteration."""
return PriceAssumptions(
oil_price_usd_bbl=samples['oil_price'][index],
gas_price_usd_mmbtu=samples['gas_price'][index],
ngl_price_usd_bbl=self.base_prices.ngl_price_usd_bbl,
annual_escalation_pct=self.base_prices.annual_escalation_pct
)
def _modify_capex(self, samples: Dict, index: int) -> CapexSchedule:
"""Create modified CAPEX for iteration."""
mult = samples['capex_mult'][index]
return replace(self.capex,
schedule={y: c * mult for y, c in self.capex.schedule.items()})
def _modify_opex(self, samples: Dict, index: int) -> OpexAssumptions:
"""Create modified OPEX for iteration."""
mult = samples['opex_mult'][index]
return replace(self.opex,
fixed_opex_usd_year=self.opex.fixed_opex_usd_year * mult,
variable_opex_usd_boe=self.opex.variable_opex_usd_boe * mult)
def _modify_production(self, samples: Dict, index: int) -> List[ProductionForecast]:
"""Create modified production forecast for iteration."""
mult = samples['prod_mult'][index]
return [
ProductionForecast(
year=p.year,
oil_mbbls=p.oil_mbbls * mult,
gas_mmcf=p.gas_mmcf * mult,
ngl_mbbls=p.ngl_mbbls * mult,
water_mbbls=p.water_mbbls
)
for p in self.production
]
def quick_simulation(self,
oil_range_pct: float = 0.30,
cost_range_pct: float = 0.20,
prod_range_pct: float = 0.15,
n_iterations: int = 1000) -> MonteCarloResult:
"""
Run quick Monte Carlo with triangular distributions.
Args:
oil_range_pct: Price range as percent of base (e.g., 0.30 = ±30%)
cost_range_pct: Cost range as percent of base
prod_range_pct: Production range as percent of base
n_iterations: Number of iterations
Returns:
MonteCarloResult with P10/P50/P90 values
"""
base_oil = self.base_prices.oil_price_usd_bbl
base_gas = self.base_prices.gas_price_usd_mmbtu
config = MonteCarloConfig(
n_iterations=n_iterations,
oil_price=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=base_oil * (1 - oil_range_pct),
mode_value=base_oil,
max_value=base_oil * (1 + oil_range_pct)
),
gas_price=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=base_gas * (1 - oil_range_pct),
mode_value=base_gas,
max_value=base_gas * (1 + oil_range_pct)
),
capex_multiplier=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=1 - cost_range_pct * 0.5, # CAPEX rarely comes in under
mode_value=1.0,
max_value=1 + cost_range_pct
),
opex_multiplier=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=1 - cost_range_pct,
mode_value=1.0,
max_value=1 + cost_range_pct
),
production_multiplier=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=1 - prod_range_pct,
mode_value=1.0,
max_value=1 + prod_range_pct * 0.5 # Production rarely exceeds forecast
)
)
return self.run_simulation(config)
def value_at_risk(self, result: MonteCarloResult,
confidence: float = 0.95) -> float:
"""
Calculate Value at Risk (VaR) for NPV.
Args:
result: Monte Carlo simulation result
confidence: Confidence level (e.g., 0.95 for 95%)
Returns:
VaR value (maximum expected loss at confidence level)
"""
percentile = (1 - confidence) * 100
return np.percentile(result.npv_values, percentile)
def conditional_value_at_risk(self, result: MonteCarloResult,
confidence: float = 0.95) -> float:
"""
Calculate Conditional VaR (Expected Shortfall).
Args:
result: Monte Carlo simulation result
confidence: Confidence level
Returns:
CVaR value (average loss in worst cases beyond VaR)
"""
var = self.value_at_risk(result, confidence)
return np.mean(result.npv_values[result.npv_values <= var])
Report Generator
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from pathlib import Path
class NPVReportGenerator:
"""Generate interactive HTML reports for NPV analysis."""
def __init__(self, calculator: NPVCalculator,
scenario_analyzer: ScenarioAnalyzer = None,
monte_carlo_result: MonteCarloResult = None):
"""Initialize report generator."""
self.calculator = calculator
self.analyzer = scenario_analyzer
self.mc_result = monte_carlo_result
def generate_report(self, output_path: Path, project_name: str = "Project"):
"""Generate comprehensive NPV analysis report with Monte Carlo."""
# Determine subplot layout based on Monte Carlo availability
if self.mc_result:
fig = make_subplots(
rows=4, cols=2,
subplot_titles=(
'Annual Cash Flows',
'Cumulative Cash Flow',
'Price Scenario NPV',
'Sensitivity Analysis',
'Monte Carlo NPV Distribution',
'P10/P50/P90 Summary',
'Production Profile',
'Revenue Breakdown'
),
vertical_spacing=0.08,
specs=[[{}, {}], [{}, {}], [{}, {}], [{}, {}]]
)
else:
fig = make_subplots(
rows=3, cols=2,
subplot_titles=(
'Annual Cash Flows',
'Cumulative Cash Flow',
'Price Scenario NPV',
'Sensitivity Analysis',
'Production Profile',
'Revenue Breakdown'
),
vertical_spacing=0.10
)
cf_df = self.calculator.to_dataframe()
# Annual cash flows
fig.add_trace(
go.Bar(x=cf_df['year'], y=cf_df['atcf']/1e6,
name='After-Tax CF ($MM)', marker_color='green'),
row=1, col=1
)
fig.add_trace(
go.Bar(x=cf_df['year'], y=-cf_df['capex']/1e6,
name='CAPEX ($MM)', marker_color='red'),
row=1, col=1
)
# Cumulative cash flow
fig.add_trace(
go.Scatter(x=cf_df['year'], y=cf_df['cumulative']/1e6,
name='Cumulative CF', fill='tozeroy',
line=dict(color='blue')),
row=1, col=2
)
fig.add_hline(y=0, line_dash='dash', line_color='gray', row=1, col=2)
# Price scenarios
if self.analyzer:
scenarios = self.analyzer.run_price_scenarios()
fig.add_trace(
go.Bar(x=scenarios['scenario'], y=scenarios['npv_mm'],
name='NPV by Scenario',
marker_color=['#d62728', '#2ca02c', '#1f77b4']),
row=2, col=1
)
# Tornado chart
tornado = self.analyzer.tornado_chart_data(self.calculator.prices)
for _, row_data in tornado.iterrows():
fig.add_trace(
go.Bar(
y=[row_data['variable']],
x=[row_data['high_npv'] - row_data['base_npv']],
orientation='h',
name=f"{row_data['variable']} High",
…(truncated)