tegwick
277c2d05ca
Add SQLite-backed cost ledger with import/summary CLI, cron-friendly evaluate command with federation emit, and dogfood evidence artefact generation from fixture CSVs.
56 lines
No EOL
1.9 KiB
Python
56 lines
No EOL
1.9 KiB
Python
"""Scheduled runway evaluation from ledger data."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from fin_hub.coupling.canon import emit_viability_alert
|
|
from fin_hub.coupling.dev_hub import emit_resource_pressure
|
|
from fin_hub.services.alerts import evaluate_budget_alerts
|
|
from fin_hub.services.ledger import get_opening_balance, monthly_burn_series
|
|
from fin_hub.services.runway import compute_runway
|
|
|
|
|
|
def evaluate_runway(
|
|
*,
|
|
ledger_path: Path,
|
|
opening_balance: float | None = None,
|
|
alert_threshold_months: float = 3.0,
|
|
currency: str = "EUR",
|
|
allocated: float | None = None,
|
|
spent: float | None = None,
|
|
emit: bool = False,
|
|
api_base: str | None = None,
|
|
) -> dict:
|
|
stored_balance, stored_currency = get_opening_balance(ledger_path)
|
|
balance = opening_balance if opening_balance is not None else stored_balance
|
|
if balance is None:
|
|
raise ValueError("opening balance required — use ledger set-balance or --balance")
|
|
|
|
effective_currency = currency or stored_currency
|
|
burns = monthly_burn_series(ledger_path=ledger_path, currency=effective_currency)
|
|
runway = compute_runway(
|
|
current_balance=balance,
|
|
monthly_burns=burns,
|
|
alert_threshold_months=alert_threshold_months,
|
|
currency=effective_currency,
|
|
)
|
|
alerts = evaluate_budget_alerts(runway=runway, allocated=allocated, spent=spent)
|
|
|
|
report: dict = {
|
|
"runway": runway.as_dict(),
|
|
"alerts": [alert.as_dict() for alert in alerts],
|
|
"monthly_burns": burns,
|
|
"ledger_path": str(ledger_path.resolve()),
|
|
}
|
|
|
|
if emit:
|
|
report["dev_hub"] = emit_resource_pressure(alerts, api_base=api_base)
|
|
report["canon"] = emit_viability_alert(runway, api_base=api_base)
|
|
|
|
return report
|
|
|
|
|
|
def evaluate_runway_json(**kwargs) -> str:
|
|
return json.dumps(evaluate_runway(**kwargs), indent=2, default=str) |