You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
202 lines
6.9 KiB
Python
202 lines
6.9 KiB
Python
"""
|
|
Collected formulas and helpers extracted from module algorithms.
|
|
|
|
This file consolidates the core mathematical/financial formulas used across:
|
|
- `Eac.py` (NPV, PMT, discounted projections, EAC calculation)
|
|
- `insert_actual_data.py` (aggregation formulas, man-hour conversion)
|
|
- `Prediksi.py` (future value / fv wrappers)
|
|
|
|
### Prediction Logic Summary
|
|
| Category | Logic Type | Formula Basis |
|
|
| :--- | :--- | :--- |
|
|
| **CM Labor** | **Reliability-Based** | `Failures x 3.0 x 1.0 x ManPowerRate` |
|
|
| **CM Other** | **Reliability-Based** | `Failures x CostPerFailure (from Production SQL)` |
|
|
| **PM / OH / PDM** | **Last Scenario** | `Value from Last Actual Year` (Carry Forward) |
|
|
| **Total Risk Cost** | **Aggregated** | `Sum of above + Asset Criticality Multiplier` |
|
|
|
|
Keep these functions pure and well-documented to make debugging and
|
|
comparisons easier.
|
|
"""
|
|
from typing import Iterable, List, Sequence
|
|
import math
|
|
import numpy as np
|
|
try:
|
|
import numpy_financial as npf
|
|
except Exception: # numpy_financial may not be installed in all envs
|
|
npf = None
|
|
|
|
|
|
def npv_from_cash_flows(cash_flows: Sequence[float], rate: float, offset: int = 0) -> float:
|
|
"""Compute discounted sum (NPV) of `cash_flows` at `rate` with optional time offset.
|
|
|
|
Formula used in `Eac.py`:
|
|
NPV = Σ [Ct / (1 + r)^(offset + t)]
|
|
where t is 1-based index of flows in `cash_flows`.
|
|
|
|
Args:
|
|
cash_flows: sequence of cash flows (Ct) for consecutive periods.
|
|
rate: periodic discount/inflation rate (e.g., 0.05 for 5%).
|
|
offset: integer number of periods to offset (e.g., last_seq from actuals).
|
|
|
|
Returns:
|
|
Discounted present value as float.
|
|
"""
|
|
if not cash_flows:
|
|
return 0.0
|
|
total = 0.0
|
|
for i, value in enumerate(cash_flows):
|
|
exp = offset + i + 1
|
|
total += float(value) / ((1.0 + rate) ** exp)
|
|
return float(total)
|
|
|
|
|
|
def pmt_formula(rate: float, nper: int, pv: float) -> float:
|
|
"""Compute level payment PMT for given rate, periods and present value.
|
|
|
|
Uses the standard annuity payment formula (same as numpy_financial.pmt but
|
|
returned value uses same sign-convention as mathematical formula):
|
|
|
|
PMT = PV * [r (1+r)^n] / [(1+r)^n - 1]
|
|
|
|
Special-case when rate == 0 (equal division): PMT = PV / nper
|
|
|
|
Note: many code places use sign flips (e.g. `-npf.pmt(...)`) because of
|
|
numpy_financial's cashflow sign policy. Use whichever sign convention your
|
|
caller expects.
|
|
|
|
Args:
|
|
rate: periodic rate (decimal)
|
|
nper: number of periods (int)
|
|
pv: present value to amortize
|
|
|
|
Returns:
|
|
payment per period (float)
|
|
"""
|
|
if nper == 0:
|
|
raise ValueError("nper must be > 0")
|
|
if abs(rate) < 1e-12:
|
|
return float(pv) / float(nper)
|
|
r = float(rate)
|
|
n = int(nper)
|
|
numerator = pv * r * ((1 + r) ** n)
|
|
denominator = ((1 + r) ** n) - 1
|
|
return float(numerator / denominator)
|
|
|
|
|
|
def pmt(rate: float, nper: int, pv: float) -> float:
|
|
"""Wrapper to compute PMT using `numpy_financial` when available, else fall back to formula.
|
|
|
|
Returns the value that `numpy_financial.pmt` would return (i.e. typically
|
|
negative for positive PV by that library's sign convention). The caller
|
|
can negate the result to get a positive payment if desired (as seen in
|
|
`Eac.py` where they do `-npf.pmt(...)`).
|
|
"""
|
|
if npf is not None:
|
|
return float(npf.pmt(rate, nper, pv))
|
|
# numpy_financial returns payments with sign convention; our fallback
|
|
# returns payment with the same absolute magnitude but as negative (to
|
|
# match npf behavior for positive PV) so callers that negate will get the
|
|
# same sign semantics.
|
|
res = pmt_formula(rate, nper, pv)
|
|
return -res
|
|
|
|
|
|
def fv(rate: float, nper: int, pmt: float, pv: float) -> float:
|
|
"""Future value wrapper to `numpy_financial.fv` when available.
|
|
|
|
Used in `Prediksi.py` as `npf.fv(rate, nper + i, pmt, pv)` to compute
|
|
forecasting values.
|
|
"""
|
|
if npf is not None:
|
|
return float(npf.fv(rate, nper, pmt, pv))
|
|
# fallback: compute iteratively
|
|
r = float(rate)
|
|
n = int(nper)
|
|
fv_val = -pv
|
|
# apply periods
|
|
for _ in range(n):
|
|
fv_val = fv_val * (1 + r) + (-pmt)
|
|
return float(fv_val)
|
|
|
|
|
|
def discounted_projection_sum(values: Sequence[float], inflation_rate: float, last_seq: int = 0) -> float:
|
|
"""Discount a sequence of projected `values` using `inflation_rate` with an initial offset `last_seq`.
|
|
|
|
This mirrors the calculation in `Eac.py` where projected flows are discounted
|
|
starting at `last_seq + 1`.
|
|
"""
|
|
return npv_from_cash_flows(values, inflation_rate, offset=last_seq)
|
|
|
|
|
|
def rc_labor_cost(interval: float, labor_time: float, labor_human: float, man_hour_rate: float) -> float:
|
|
"""Compute labor cost for corrective/pm/oh rows.
|
|
|
|
Formula from `Prediksi.py` / `insert_actual_data.py`:
|
|
rc_xxx_labor_cost = (interval * labor_time * labor_human * man_hour_rate)
|
|
"""
|
|
return float(interval) * float(labor_time) * float(labor_human) * float(man_hour_rate)
|
|
|
|
|
|
def rc_lost_cost(raw_loss_output_mw: float, raw_loss_output_price: float, cm_interval: float) -> float:
|
|
"""Compute lost production cost as in `Prediksi.py`.
|
|
|
|
Formula used:
|
|
rc_lost_cost = (raw_loss_output_MW * raw_loss_output_price * raw_cm_interval) * 1000
|
|
The multiplication by 1000 appears in source and is preserved here.
|
|
"""
|
|
return float(raw_loss_output_mw) * float(raw_loss_output_price) * float(cm_interval) * 1000.0
|
|
|
|
|
|
def rc_total_cost(**components) -> float:
|
|
"""Sum all provided components into a total RC cost.
|
|
|
|
The `Eac.py` / `Prediksi.py` code builds a `rc_total_cost` as the sum of many
|
|
parts; this helper simply sums values passed as keyword args (coalescing
|
|
None to 0.0) to make testing/comparison easier.
|
|
"""
|
|
total = 0.0
|
|
for k, v in components.items():
|
|
try:
|
|
total += 0.0 if v is None else float(v)
|
|
except Exception:
|
|
# ignore non-numeric components
|
|
continue
|
|
return float(total)
|
|
|
|
|
|
def seconds_to_hours(seconds: float, ndigits: int = 2) -> float:
|
|
"""Convert seconds to hours and round to `ndigits` decimals.
|
|
|
|
This mirrors usage like:
|
|
ROUND(SUM(EXTRACT(EPOCH FROM (actfinish - actstart)) / 3600), 2)
|
|
in SQL queries seen in `insert_actual_data.py`.
|
|
"""
|
|
hours = float(seconds) / 3600.0
|
|
return float(round(hours, ndigits))
|
|
|
|
|
|
def man_count_from_labtrans(count: int) -> int:
|
|
"""Return man_count as per SQL CASE: if count == 0 -> 3 else count.
|
|
|
|
Implemented to match the SQL "CASE WHEN COUNT(b.laborcode) = 0 THEN 3 ELSE COUNT(b.laborcode)".
|
|
"""
|
|
try:
|
|
c = int(count)
|
|
except Exception:
|
|
c = 0
|
|
return 3 if c == 0 else c
|
|
|
|
|
|
__all__ = [
|
|
"npv_from_cash_flows",
|
|
"pmt_formula",
|
|
"pmt",
|
|
"fv",
|
|
"discounted_projection_sum",
|
|
"rc_labor_cost",
|
|
"rc_lost_cost",
|
|
"rc_total_cost",
|
|
"seconds_to_hours",
|
|
"man_count_from_labtrans",
|
|
]
|