CIzz22
/
be-lcca
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update run2 plant - create new function liner generate for B&D Cost

Browse Source
main
ariwahyunahar 3 weeks ago
parent b9df675099
commit ef07faf9a0
1 changed files with 105 additions and 24 deletions
Show Stats Download Patch File Download Diff File

129
src/modules/plant/run2.py
Unescape Escape View File

@ -70,6 +70,67 @@ def hitung_pv(rate, nper, fv):
def hitung_irr(cashflows: list):
return npf.irr(cashflows)
def getproyeksilinier(years, values, iterations=30, target_years=None):
"""
Jika target_years diberikan (list[int]), fungsi mengembalikan prediksi untuk tahun-tahun tersebut.
Jika target_years None, perilaku lama tetap: memproyeksikan dari max(years)+1 sepanjang (iterations - len(years)).
Catatan:
- Jika data historis < 2 titik, fallback pakai nilai terakhir (atau 0).
- Jika semua year sama (degenerate), fallback juga.
"""
if len(years) != len(values):
raise ValueError("Panjang years dan values harus sama")
# bersihkan pasangan (year, value) yang year-nya None
pairs = [(int(y), float(v)) for y, v in zip(years, values) if y is not None]
if not pairs:
# tidak ada data sama sekali
if target_years is None:
return {}
return {int(y): 0.0 for y in target_years}
years_clean = [p[0] for p in pairs]
values_clean = [p[1] for p in pairs]
# fallback kalau data tidak cukup untuk regresi
if len(set(years_clean)) < 2 or len(values_clean) < 2:
last_val = float(values_clean[-1]) if values_clean else 0.0
if target_years is None:
# perilaku lama
n_hist = len(years_clean)
if iterations <= n_hist:
return {}
start_year = max(years_clean) + 1
n_projection = iterations - n_hist
return {start_year + i: last_val for i in range(n_projection)}
return {int(y): last_val for y in target_years}
# regresi linier y = a*x + b
x = np.array(years_clean, dtype=float)
y = np.array(values_clean, dtype=float)
a, b = np.polyfit(x, y, 1)
def _predict(yr: int) -> float:
v = float(a * yr + b)
# optional: kalau tidak boleh negatif, clamp
return max(0.0, v)
# mode target_years: prediksi untuk tahun tertentu
if target_years is not None:
return {int(yr): _predict(int(yr)) for yr in target_years}
# mode lama: generate dari tahun setelah histori
n_hist = len(years_clean)
if iterations <= n_hist:
raise ValueError(
f"iterations ({iterations}) harus lebih besar dari jumlah data historis ({n_hist})"
)
start_year = max(years_clean) + 1
n_projection = iterations - n_hist
return {start_year + i: _predict(start_year + i) for i in range(n_projection)}
def main():
connections = get_connection()
@ -169,6 +230,41 @@ def main():
col_names = [desc[0] for desc in cur.description]
rows = cur.fetchall()
# ============================================================
# PROYEKSI LINIER untuk COST BD berdasarkan histori (is_actual=1)
# ============================================================
hist_years_om, hist_vals_om = [], []
hist_years_pm, hist_vals_pm = [], []
hist_years_bd, hist_vals_bd = [], []
projection_years = []
for r in rows:
d = dict(zip(col_names, r))
yr = d.get("tahun")
if yr is None:
continue
yr = int(yr)
if d.get("is_actual") == 1:
# ambil histori (boleh 0 kalau null)
hist_years_om.append(yr)
hist_vals_om.append(validate_number(d.get("cost_bd_om")))
hist_years_pm.append(yr)
hist_vals_pm.append(validate_number(d.get("cost_bd_pm_nonmi")))
hist_years_bd.append(yr)
hist_vals_bd.append(validate_number(d.get("cost_bd_bd")))
else:
# tahun-tahun projection yang ingin diprediksi
projection_years.append(yr)
# buat mapping prediksi per tahun untuk masing-masing komponen cost BD
proj_cost_bd_om = getproyeksilinier(hist_years_om, hist_vals_om, target_years=projection_years)
proj_cost_bd_pm = getproyeksilinier(hist_years_pm, hist_vals_pm, target_years=projection_years)
proj_cost_bd_bd = getproyeksilinier(hist_years_bd, hist_vals_bd, target_years=projection_years)
print(f"Jumlah baris yang akan di-update: {len(rows)}")
# 2. Siapkan data untuk bulk UPDATE
@ -317,6 +413,7 @@ def main():
data = dict(zip(col_names, row))
seq = data["seq"] # primary key / unique key untuk WHERE
yr = int(data["tahun"]) if data.get("tahun") is not None else None
# Ambil net_capacity_factor dan eaf dari cache berdasarkan tahun
cf_eaf = year_data_map.get(int(data["tahun"])) if data.get("tahun") is not None else None
@ -349,31 +446,15 @@ def main():
revenue_c = price_c * production_netto * 1000 / 1000000
revenue_d = price_d * production_netto * 1000 / 1000000
cost_c_fuel = fuel_consumption * harga_bahan_bakar / 1000000
# Linear prediction for cost_bd_om, cost_bd_pm_nonmi, cost_bd_bd
# Use last 2 actual values to predict next value (simple linear extrapolation)
def linear_predict(last_vals):
if len(last_vals) >= 2:
return last_vals[-1] + (last_vals[-1] - last_vals[-2])
elif len(last_vals) == 1:
return last_vals[-1]
else:
return 0
# Collect actual values up to now
if not hasattr(main, "_cost_bd_om_actuals"):
main._cost_bd_om_actuals = []
main._cost_bd_pm_nonmi_actuals = []
main._cost_bd_bd_actuals = []
# If actual, append to history
if data["is_actual"] == 1:
main._cost_bd_om_actuals.append(cost_bd_om)
main._cost_bd_pm_nonmi_actuals.append(cost_bd_pm_nonmi)
main._cost_bd_bd_actuals.append(cost_bd_bd)
# default fallback tetap pakai last value kalau tahun kosong / prediksi tidak ada
if yr is not None:
cost_bd_om = proj_cost_bd_om.get(yr, cost_bd_om)
cost_bd_pm_nonmi = proj_cost_bd_pm.get(yr, cost_bd_pm_nonmi)
cost_bd_bd = proj_cost_bd_bd.get(yr, cost_bd_bd)
else:
cost_bd_om = linear_predict(main._cost_bd_om_actuals)
cost_bd_pm_nonmi = linear_predict(main._cost_bd_pm_nonmi_actuals)
cost_bd_bd = linear_predict(main._cost_bd_bd_actuals)
cost_bd_om = cost_bd_om
cost_bd_pm_nonmi = cost_bd_pm_nonmi
cost_bd_bd = cost_bd_bd
net_capacity_factor = net_capacity_factor_v
eaf = eaf_v

Write Preview
Loading…
Cancel
Save