| name | npv-analyzer |
| description | Perform NPV analysis and economic evaluation for oil & gas assets. Use for cash flow modeling, price scenario analysis, Monte Carlo simulation, P10/P50/P90 probabilistic analysis, working interest calculations, and financial metrics (IRR, payback, NPV) for field development projects. |
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"
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
annual_escalation_pct: float = 2.0
price_deck: Optional[Dict[int, Tuple[float, float]]] = None
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
net_revenue_interest: float = 0.875
royalty_rate: float = 0.125
state_severance_tax: float = 0.0
federal_income_tax: float = 0.21
ad_valorem_tax: float = 0.0
cost_recovery_limit: float = 1.0
profit_oil_share: float = 1.0
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
completion_per_well: float = 0.0
facilities: float = 0.0
subsea_equipment: float = 0.0
pipeline: float = 0.0
other_capex: float = 0.0
schedule: Dict[int, float] = field(default_factory=dict)
@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
variable_opex_usd_boe: float = 0.0
workover_budget: float = 0.0
insurance: float = 0.0
g_and_a: float = 0.0
abandonment_cost: float = 0.0
abandonment_year: Optional[int] = None
@dataclass
class ProductionForecast:
"""Production forecast by year."""
year: int
oil_mbbls: float = 0.0
gas_mmcf: float = 0.0
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
@dataclass
class CashFlowYear:
"""Annual cash flow calculation."""
year: int
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
capex: float = 0.0
opex: float = 0.0
severance_tax: float = 0.0
ad_valorem_tax: float = 0.0
abandonment: float = 0.0
taxable_income: float = 0.0
income_tax: float = 0.0
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
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)
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)
)
cf.gross_oil_revenue = prod.oil_mbbls * oil_price * 1000
cf.gross_gas_revenue = prod.gas_mmbtu * gas_price / 1000
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)
cf.net_revenue = cf.total_gross_revenue * self.fiscal.net_revenue_factor()
cf.capex = self.capex.schedule.get(prod.year, 0.0) * 1e6
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
)
if (self.opex.abandonment_year and
prod.year == self.opex.abandonment_year):
cf.abandonment = self.opex.abandonment_cost
cf.severance_tax = (cf.total_gross_revenue *
self.fiscal.working_interest *
self.fiscal.state_severance_tax)
cf.ad_valorem_tax = (cf.total_gross_revenue *
self.fiscal.working_interest *
self.fiscal.ad_valorem_tax)
cf.before_tax_cash_flow = (cf.net_revenue -
cf.capex -
cf.opex -
cf.severance_tax -
cf.ad_valorem_tax -
cf.abandonment)
cf.taxable_income = max(0, cf.before_tax_cash_flow)
cf.income_tax = cf.taxable_income * self.fiscal.federal_income_tax
cf.after_tax_cash_flow = cf.before_tax_cash_flow - cf.income_tax
cumulative += cf.after_tax_cash_flow
cf.cumulative_cash_flow = cumulative
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
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
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:
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.
"""
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 = []
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
elif variable == 'opex':
base_value = 1.0
elif variable == 'discount_rate':
base_value = self.discount_rate
else:
raise ValueError(f"Unknown variable: {variable}")
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:
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
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
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 = []
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
max_value: float = 0.0
mean: Optional[float] = None
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:
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
oil_price: Optional[DistributionParams] = None
gas_price: Optional[DistributionParams] = None
capex_multiplier: Optional[DistributionParams] = None
opex_multiplier: Optional[DistributionParams] = None
production_multiplier: Optional[DistributionParams] = None
discount_rate: Optional[DistributionParams] = None
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
p10_npv: float = 0.0
p50_npv: float = 0.0
p90_npv: float = 0.0
p10_irr: float = 0.0
p50_irr: float = 0.0
p90_irr: float = 0.0
mean_npv: float = 0.0
std_npv: float = 0.0
prob_positive_npv: float = 0.0
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
samples = self._generate_samples(config, n, rng)
npv_values = np.zeros(n)
irr_values = np.zeros(n)
payback_values = np.zeros(n)
for i in range(n):
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]
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')
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,
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
)
)
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."""
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()
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
)
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)
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 = 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",
marker_color='green',
showlegend=False
),
row=2, col=2
)
fig.add_trace(
go.Bar(
y=[row_data['variable']],
x=[row_data['low_npv'] - row_data['base_npv']],
orientation='h',
name=f"{row_data['variable']} Low",
marker_color='red',
showlegend=False
),
row=2, col=2
)
mc_row = 3 if self.mc_result else None
prod_row = 4 if self.mc_result else 3
if self.mc_result:
fig.add_trace(
go.Histogram(
x=self.mc_result.npv_values / 1e6,
nbinsx=50,
name='NPV Distribution',
marker_color='steelblue',
opacity=0.7
),
row=3, col=1
)
for pct, val, color in [
('P10', self.mc_result.p10_npv, 'red'),
('P50', self.mc_result.p50_npv, 'orange'),
('P90', self.mc_result.p90_npv, 'green')
]:
fig.add_vline(
x=val / 1e6, line_dash='dash', line_color=color,
annotation_text=pct, row=3, col=1
)
fig.add_trace(
go.Bar(
x=['P10', 'P50', 'P90'],
y=[self.mc_result.p10_npv / 1e6,
self.mc_result.p50_npv / 1e6,
self.mc_result.p90_npv / 1e6],
marker_color=['red', 'orange', 'green'],
name='Percentiles'
),
row=3, col=2
)
mc_summary = (
f"<b>Monte Carlo Results</b><br>"
f"Iterations: {len(self.mc_result.npv_values):,}<br>"
f"Mean NPV: ${self.mc_result.mean_npv/1e6:.1f}MM<br>"
f"Std Dev: ${self.mc_result.std_npv/1e6:.1f}MM<br>"
f"P(NPV>0): {self.mc_result.prob_positive_npv*100:.1f}%"
)
fig.add_annotation(
text=mc_summary,
xref='paper', yref='paper',
x=1.02, y=0.50,
showarrow=False,
font=dict(size=11),
align='left',
bgcolor='lightyellow',
bordercolor='gray'
)
prod_df = pd.DataFrame([
{'year': p.year, 'oil': p.oil_mbbls, 'gas': p.gas_mmcf/6, 'ngl': p.ngl_mbbls}
for p in self.calculator.production
])
fig.add_trace(
go.Bar(x=prod_df['year'], y=prod_df['oil'],
name='Oil (Mbbls)', marker_color='green'),
row=prod_row, col=1
)
fig.add_trace(
go.Bar(x=prod_df['year'], y=prod_df['gas'],
name='Gas (BOE Mbbls)', marker_color='red'),
row=prod_row, col=1
)
cash_flows = self.calculator.calculate_cash_flows()
total_oil_rev = sum(cf.gross_oil_revenue for cf in cash_flows)
total_gas_rev = sum(cf.gross_gas_revenue for cf in cash_flows)
total_ngl_rev = sum(cf.gross_ngl_revenue for cf in cash_flows)
fig.add_trace(
go.Pie(
labels=['Oil', 'Gas', 'NGL'],
values=[total_oil_rev, total_gas_rev, total_ngl_rev],
marker_colors=['green', 'red', 'orange']
),
row=prod_row, col=2
)
summary = self.calculator.summary()
summary_text = (
f"<b>Economic Summary</b><br>"
f"NPV (10%): ${summary['npv_mm']:.1f}MM<br>"
f"IRR: {summary['irr_pct']:.1f}%<br>"
f"Payback: {summary['payback_years']:.1f} years<br>"
f"PI: {summary['profitability_index']:.2f}"
)
fig.add_annotation(
text=summary_text,
xref='paper', yref='paper',
x=1.02, y=0.98,
showarrow=False,
font=dict(size=12),
align='left',
bgcolor='white',
bordercolor='gray'
)
fig.update_layout(
height=1400 if self.mc_result else 1100,
title_text=f"{project_name} - NPV Analysis" + (" (Monte Carlo)" if self.mc_result else ""),
showlegend=True,
barmode='relative'
)
output_path.parent.mkdir(parents=True, exist_ok=True)
fig.write_html(str(output_path))
return output_path
YAML Configuration
Project Configuration
metadata:
project_name: "Lower Tertiary Development"
analyst: "Engineering Team"
date: "2024-01-15"
version: "1.0"
economics:
discount_rate: 0.10
base_year: 2024
project_life: 25
prices:
scenario: mid
oil_price_usd_bbl: 70.0
gas_price_usd_mmbtu: 3.50
ngl_price_usd_bbl: 35.0
annual_escalation_pct: 2.0
fiscal:
working_interest: 0.50
net_revenue_interest: 0.875
royalty_rate: 0.125
state_severance_tax: 0.0
federal_income_tax: 0.21
ad_valorem_tax: 0.0
capex:
drilling_per_well: 120.0
completion_per_well: 40.0
facilities: 500.0
subsea_equipment: 300.0
pipeline: 150.0
other_capex: 50.0
schedule:
2024: 200.0
2025: 600.0
2026: 360.0
opex:
fixed_opex_usd_year: 25000000
variable_opex_usd_boe: 8.0
workover_budget: 5000000
insurance: 3000000
g_and_a: 2000000
abandonment_cost: 150000000
abandonment_year: 2049
production:
source: "file"
file_path: "data/production_forecast.csv"
output:
cash_flow_csv: "data/results/cash_flows.csv"
summary_json: "data/results/summary.json"
report_html: "reports/npv_analysis.html"
Scenario Comparison
scenarios:
- name: "Base Case"
oil_price: 70
gas_price: 3.50
capex_multiplier: 1.0
- name: "Low Price"
oil_price: 50
gas_price: 2.50
capex_multiplier: 1.0
- name: "High Price"
oil_price: 90
gas_price: 5.00
capex_multiplier: 1.0
- name: "Cost Overrun"
oil_price: 70
gas_price: 3.50
capex_multiplier: 1.25
- name: "Fast Payout"
oil_price: 80
gas_price: 4.00
capex_multiplier: 0.90
analysis:
run_sensitivity: true
sensitivity_range_pct: 0.30
tornado_chart: true
output:
comparison_csv: "data/results/scenario_comparison.csv"
report_html: "reports/scenario_comparison.html"
Monte Carlo Configuration
metadata:
project_name: "Lower Tertiary Development"
analyst: "Risk Analysis Team"
date: "2024-01-15"
version: "1.0"
monte_carlo:
n_iterations: 10000
random_seed: 42
oil_price:
distribution: triangular
min_value: 50.0
mode_value: 70.0
max_value: 100.0
gas_price:
distribution: triangular
min_value: 2.50
mode_value: 3.50
max_value: 6.00
capex_multiplier:
distribution: pert
min_value: 0.95
mode_value: 1.00
max_value: 1.30
opex_multiplier:
distribution: triangular
min_value: 0.85
mode_value: 1.00
max_value: 1.20
production_multiplier:
distribution: pert
min_value: 0.70
mode_value: 1.00
max_value: 1.10
output:
npv_distribution_csv: "data/results/npv_distribution.csv"
percentiles_json: "data/results/percentiles.json"
report_html: "reports/monte_carlo_analysis.html"
CLI Usage
python -m npv_analyzer run --config config/npv_analysis.yaml
python -m npv_analyzer calculate \
--production data/forecast.csv \
--oil-price 70 \
--gas-price 3.50 \
--discount-rate 0.10
python -m npv_analyzer scenarios --config config/scenario_comparison.yaml
python -m npv_analyzer breakeven --config config/npv_analysis.yaml --commodity oil
python -m npv_analyzer sensitivity --config config/npv_analysis.yaml --range 0.30
python -m npv_analyzer montecarlo --config config/monte_carlo_analysis.yaml
python -m npv_analyzer montecarlo --config config/npv_analysis.yaml \
--iterations 5000 \
--oil-range 0.30 \
--cost-range 0.20
python -m npv_analyzer percentiles --config config/monte_carlo_analysis.yaml
Usage Examples
Example 1: Simple NPV Calculation
from npv_analyzer import (
NPVCalculator, ProductionForecast, PriceAssumptions,
FiscalTerms, CapexSchedule, OpexAssumptions
)
production = [
ProductionForecast(year=2027, oil_mbbls=3000, gas_mmcf=2000),
ProductionForecast(year=2028, oil_mbbls=8000, gas_mmcf=5000),
ProductionForecast(year=2029, oil_mbbls=10000, gas_mmcf=6500),
ProductionForecast(year=2030, oil_mbbls=9000, gas_mmcf=6000),
ProductionForecast(year=2031, oil_mbbls=7500, gas_mmcf=5000),
]
prices = PriceAssumptions(oil_price_usd_bbl=70, gas_price_usd_mmbtu=3.50)
fiscal = FiscalTerms(working_interest=0.50, net_revenue_interest=0.875)
capex = CapexSchedule(schedule={2024: 200, 2025: 600, 2026: 360})
opex = OpexAssumptions(fixed_opex_usd_year=25e6, variable_opex_usd_boe=8.0)
calc = NPVCalculator(production, prices, fiscal, capex, opex, discount_rate=0.10)
print(f"NPV @ 10%: ${calc.npv()/1e6:.1f}MM")
print(f"IRR: {calc.irr()*100:.1f}%")
print(f"Payback: {calc.payback_period():.1f} years")
Example 2: Scenario Analysis
from npv_analyzer import ScenarioAnalyzer, NPVReportGenerator
analyzer = ScenarioAnalyzer(production, fiscal, capex, opex)
scenarios = analyzer.run_price_scenarios()
print("\nPrice Scenario Results:")
print(scenarios[['scenario', 'npv_mm', 'irr_pct', 'payback_years']])
tornado = analyzer.tornado_chart_data(prices)
print("\nSensitivity Ranking:")
print(tornado[['variable', 'range']])
reporter = NPVReportGenerator(calc, analyzer)
reporter.generate_report(
Path("reports/npv_report.html"),
project_name="Lower Tertiary Development"
)
Example 3: Monte Carlo Simulation
from npv_analyzer import (
MonteCarloSimulator, MonteCarloConfig, DistributionParams,
DistributionType, NPVReportGenerator
)
mc_sim = MonteCarloSimulator(
production=production,
base_prices=prices,
fiscal=fiscal,
capex=capex,
opex=opex,
base_discount_rate=0.10
)
result = mc_sim.quick_simulation(
oil_range_pct=0.30,
cost_range_pct=0.20,
prod_range_pct=0.15,
n_iterations=5000
)
print(f"\nMonte Carlo Results ({len(result.npv_values):,} iterations):")
print(f"P10 NPV: ${result.p10_npv/1e6:.1f}MM (pessimistic)")
print(f"P50 NPV: ${result.p50_npv/1e6:.1f}MM (median)")
print(f"P90 NPV: ${result.p90_npv/1e6:.1f}MM (optimistic)")
print(f"Probability of positive NPV: {result.prob_positive_npv*100:.1f}%")
custom_config = MonteCarloConfig(
n_iterations=10000,
random_seed=42,
oil_price=DistributionParams(
distribution=DistributionType.TRIANGULAR,
min_value=50, mode_value=70, max_value=100
),
capex_multiplier=DistributionParams(
distribution=DistributionType.PERT,
min_value=0.95, mode_value=1.0, max_value=1.30
),
production_multiplier=DistributionParams(
distribution=DistributionType.LOGNORMAL,
min_value=0.6, max_value=1.15, mean=1.0, std_dev=0.15
)
)
result = mc_sim.run_simulation(custom_config)
var_95 = mc_sim.value_at_risk(result, confidence=0.95)
cvar_95 = mc_sim.conditional_value_at_risk(result, confidence=0.95)
print(f"\nRisk Metrics:")
print(f"VaR (95%): ${var_95/1e6:.1f}MM")
print(f"CVaR (95%): ${cvar_95/1e6:.1f}MM")
reporter = NPVReportGenerator(calc, analyzer, monte_carlo_result=result)
reporter.generate_report(
Path("reports/npv_montecarlo_report.html"),
project_name="Lower Tertiary Development - Risk Analysis"
)
Best Practices
Model Setup
- Use consistent units throughout (USD, bbls, MCF)
- Validate production forecasts against reservoir studies
- Document all assumptions in YAML configs
- Version control economic parameters
Sensitivity Analysis
- Always run multiple price scenarios
- Include cost and schedule sensitivities
- Identify breakeven prices for investment decisions
- Document uncertainty ranges
Reporting
- Include summary metrics prominently
- Show cash flow timing clearly
- Compare against investment criteria (hurdle rate, payback limits)
- Archive analysis with assumptions
Monte Carlo Analysis
- Use 5,000-10,000 iterations for stable P10/P50/P90
- Choose appropriate distributions (PERT for expert estimates, triangular for simple ranges)
- CAPEX typically skewed toward overruns (use asymmetric distributions)
- Production typically skewed toward underperformance
- Set random seed for reproducible results
- Calculate VaR and CVaR for risk management
- Report probability of positive NPV for investment decisions
Related Skills