tegwick
ce4f791a63
Add ai-plan ingest that stores one current booked fact per provider-account × plan × period. Subscriptions become recurring commitments; usage top-ups do not. Session-level rows are rejected.
68 lines
No EOL
2.2 KiB
Python
68 lines
No EOL
2.2 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,
|
|
list_current_commitments,
|
|
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)
|
|
commitments = list_current_commitments(
|
|
ledger_path=ledger_path, currency=effective_currency
|
|
)
|
|
if not burns:
|
|
committed = sum((float(item.amount) for item in commitments), 0.0)
|
|
if committed > 0:
|
|
burns = [committed]
|
|
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,
|
|
"commitments": [item.as_dict() for item in commitments],
|
|
"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) |