coulomb/fin-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement FIN-WP-0001 ledger, evaluate, and evidence lane (T03–T05)

Browse source 
Add SQLite-backed cost ledger with import/summary CLI, cron-friendly
evaluate command with federation emit, and dogfood evidence artefact
generation from fixture CSVs.
This commit is contained in:
tegwick 2026-07-08 22:44:07 +02:00
parent d6f88cd70a
commit 277c2d05ca
12 changed files with 760 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

7
.claude/rules/architecture.md
View file

@ -13,6 +13,9 @@ fin-specific models and services live here.
| `services/runway.py` | Runway projection from balance + monthly burn series |
| `services/alerts.py` | Threshold evaluation (runway months, budget overrun) |
| `coupling/` | FOS §9 signals: fin→dev (resource pressure), fin→ops (cost attribution), fin→canon (viability) |
| `services/ledger.py` | SQLite cost ledger with monthly rollups |
| `services/evaluate.py` | Ledger-driven runway evaluation for cron |
| `services/evidence.py` | Non-secret dogfood evidence generation |
| `cli.py` | Operator CLI (`finhub` entrypoint) |
### Data flow
@ -21,8 +24,8 @@ fin-specific models and services live here.
CSV exports → ingest parsers → cost models → runway/alerts → coupling emit → State Hub progress
```
v0.1 is manual CSV import only. Persistent ledger and HTTP read surface are
tracked in `FIN-WP-0001`.
v0.1 uses manual CSV import into a local SQLite ledger. HTTP read surface
remains deferred (`FIN-WP-0001-T06`).
## Quick Reference

11
.claude/rules/stack-and-commands.md
View file

@ -21,6 +21,17 @@ uv run finhub import-cloud tests/fixtures/cloud-costs.csv
uv run finhub import-anthropic tests/fixtures/anthropic-billing.csv
uv run finhub import-hosteurope tests/fixtures/hosteurope.csv
# CLI — ledger (SQLite at .fin-hub/ledger.db)
uv run finhub ledger set-balance 12000
uv run finhub ledger import cloud tests/fixtures/cloud-costs.csv
uv run finhub ledger summary
# CLI — scheduled evaluation (cron/systemd friendly)
uv run finhub evaluate --emit
# CLI — dogfood evidence artefact
uv run finhub evidence --seed-fixtures
# CLI — emit cross-hub signals (requires STATE_HUB_API)
uv run finhub runway --balance 12000 --monthly-burn 2100 --emit
```

1
.gitignore vendored
View file

@ -3,3 +3,4 @@ __pycache__/
*.pyc
.pytest_cache/
.env
.fin-hub/

3
README.md
View file

@ -21,6 +21,9 @@ cd /home/worsch/fin-hub
uv sync
uv run pytest
uv run finhub runway --balance 12000 --monthly-burn 2100,2200,2000
uv run finhub ledger import cloud tests/fixtures/cloud-costs.csv
uv run finhub evaluate
uv run finhub evidence --seed-fixtures
uv run finhub import-cloud tests/fixtures/cloud-costs.csv
uv run finhub ops-costs tests/fixtures/hosteurope.csv
```

41
docs/evidence/runway-dogfood-2026-07-08.md Normal file
View file

@ -0,0 +1,41 @@
# Runway Dogfood Evidence — 2026-07-08
Status: generated v0.1 (2026-07-08 20:43 UTC)
Non-secret internal dogfood artefact for RaaS managed-tier runway evidence.
Fixture CSVs under `tests/fixtures/` imported into a local SQLite ledger.
## Opening position
- Opening balance: **12,000.00 EUR**
- Alert threshold: **3.0 months**
- Ledger path: `/home/worsch/fin-hub/.fin-hub/ledger.db`
## Monthly burn (ledger rollups)
| Period | Currency | Total | Entries |
| --- | --- | ---: | ---: |
| 2026-06 | EUR | 185.30 | 5 |
| 2026-06 | USD | 1.65 | 2 |
## Runway projection
- Average monthly burn (EUR): **185.30**
- Projected runway: **64.8 months**
- Below threshold: **no**
## Alerts
- No alerts triggered.
## Federation signals
When `STATE_HUB_API` is reachable, `finhub evaluate --emit` posts:
- fin→dev resource pressure (`budget_pressure`, `runway_below_threshold`)
- fin→canon viability alert when runway is below threshold
## References
- `docs/raas-mvp-packaging.md` — managed-tier runway add-on
- `workplans/FIN-WP-0001-runway-operations-lane.md` — operational lane

4
docs/raas-mvp-packaging.md
View file

@ -54,7 +54,9 @@ instead of founder-only ops.
## First Customer Acquisition Strategy
1. **Internal dogfood** — complete Railiance restore drill + fin-hub runway evidence.
1. **Internal dogfood** — complete Railiance restore drill + fin-hub runway evidence
(`docs/evidence/runway-dogfood-2026-07-08.md`, regenerate via
`PYTHONPATH=src uv run finhub evidence --seed-fixtures`).
2. **Founder network** — 3 architecture review offers to EU technical SMEs with DSGVO pressure.
3. **Consulting wedge** — fixed-scope "sovereign ops assessment" (2–5 days) leading to managed tier.
4. **Content** — publish restore-drill case study (non-secret evidence only).

114
src/fin_hub/cli.py
View file

@ -14,9 +14,21 @@ from fin_hub.ingest.anthropic import parse_anthropic_billing_csv
from fin_hub.ingest.cloud import parse_cloud_cost_csv
from fin_hub.ingest.hosteurope import parse_hosteurope_csv
from fin_hub.services.alerts import evaluate_budget_alerts
from fin_hub.services.evaluate import evaluate_runway
from fin_hub.services.evidence import write_runway_evidence
from fin_hub.services.ledger import (
default_ledger_path,
import_csv,
ledger_stats_json,
set_opening_balance,
)
from fin_hub.services.runway import compute_runway
def _ledger_path(args: argparse.Namespace) -> Path:
return Path(args.ledger) if args.ledger else default_ledger_path()
def _cmd_import_cloud(args: argparse.Namespace) -> int:
rows = parse_cloud_cost_csv(Path(args.path))
print(json.dumps([row.__dict__ for row in rows], indent=2, default=str))
@ -82,6 +94,61 @@ def _cmd_ops_costs(args: argparse.Namespace) -> int:
return 0
def _cmd_ledger_import(args: argparse.Namespace) -> int:
result = import_csv(
Path(args.path),
args.type,
ledger_path=_ledger_path(args),
force=args.force,
)
print(json.dumps(result.as_dict(), indent=2))
return 0
def _cmd_ledger_summary(args: argparse.Namespace) -> int:
print(ledger_stats_json(ledger_path=_ledger_path(args)))
return 0
def _cmd_ledger_set_balance(args: argparse.Namespace) -> int:
set_opening_balance(_ledger_path(args), args.balance, currency=args.currency)
print(
json.dumps(
{"opening_balance": args.balance, "currency": args.currency},
indent=2,
)
)
return 0
def _cmd_evaluate(args: argparse.Namespace) -> int:
report = evaluate_runway(
ledger_path=_ledger_path(args),
opening_balance=args.balance,
alert_threshold_months=args.threshold,
currency=args.currency,
allocated=args.allocated,
spent=args.spent,
emit=args.emit,
api_base=args.api_base,
)
print(json.dumps(report, indent=2, default=str))
return 0
def _cmd_evidence(args: argparse.Namespace) -> int:
output = write_runway_evidence(
ledger_path=_ledger_path(args),
output_dir=Path(args.output_dir) if args.output_dir else None,
opening_balance=args.balance,
seed_fixtures=args.seed_fixtures,
alert_threshold_months=args.threshold,
currency=args.currency,
)
print(json.dumps({"written": str(output.resolve())}, indent=2))
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fin Hub operator CLI")
sub = parser.add_subparsers(dest="command", required=True)
@ -113,6 +180,53 @@ def build_parser() -> argparse.ArgumentParser:
ops.add_argument("path")
ops.set_defaults(func=_cmd_ops_costs)
ledger = sub.add_parser("ledger", help="Persistent SQLite cost ledger")
ledger_sub = ledger.add_subparsers(dest="ledger_command", required=True)
ledger_import = ledger_sub.add_parser("import", help="Import a cost CSV into the ledger")
ledger_import.add_argument("type", choices=["cloud", "anthropic", "hosteurope"])
ledger_import.add_argument("path")
ledger_import.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
ledger_import.add_argument("--force", action="store_true", help="Re-import even if file unchanged")
ledger_import.set_defaults(func=_cmd_ledger_import)
ledger_summary = ledger_sub.add_parser("summary", help="Monthly rollup summary from ledger")
ledger_summary.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
ledger_summary.set_defaults(func=_cmd_ledger_summary)
ledger_balance = ledger_sub.add_parser("set-balance", help="Store opening cash balance in ledger meta")
ledger_balance.add_argument("balance", type=float)
ledger_balance.add_argument("--currency", default="EUR")
ledger_balance.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
ledger_balance.set_defaults(func=_cmd_ledger_set_balance)
evaluate = sub.add_parser(
"evaluate",
help="Evaluate runway from ledger burns (cron/systemd friendly)",
)
evaluate.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
evaluate.add_argument("--balance", type=float, help="Override stored opening balance")
evaluate.add_argument("--threshold", type=float, default=3.0)
evaluate.add_argument("--currency", default="EUR")
evaluate.add_argument("--allocated", type=float)
evaluate.add_argument("--spent", type=float)
evaluate.add_argument("--emit", action="store_true", help="Emit fin→dev and fin→canon signals")
evaluate.add_argument("--api-base")
evaluate.set_defaults(func=_cmd_evaluate)
evidence = sub.add_parser("evidence", help="Write runway dogfood evidence artefact")
evidence.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
evidence.add_argument("--balance", type=float, default=12000.0)
evidence.add_argument("--threshold", type=float, default=3.0)
evidence.add_argument("--currency", default="EUR")
evidence.add_argument("--output-dir", help="Output directory (default: docs/evidence)")
evidence.add_argument(
"--seed-fixtures",
action="store_true",
help="Import tests/fixtures CSVs before generating the report",
)
evidence.set_defaults(func=_cmd_evidence)
return parser

56
src/fin_hub/services/evaluate.py Normal file
View file

@ -0,0 +1,56 @@
"""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)

159
src/fin_hub/services/evidence.py Normal file
View file

@ -0,0 +1,159 @@
"""Generate non-secret runway evidence artefacts for dogfood and RaaS."""
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
from fin_hub.services.alerts import evaluate_budget_alerts
from fin_hub.services.ledger import (
import_csv,
ledger_stats,
monthly_burn_series,
monthly_summary,
set_opening_balance,
)
from fin_hub.services.runway import compute_runway
FIXTURE_DIR = Path(__file__).resolve().parents[3] / "tests" / "fixtures"
DEFAULT_EVIDENCE_DIR = Path("docs/evidence")
def seed_fixture_ledger(
*,
ledger_path: Path,
opening_balance: float = 12000.0,
currency: str = "EUR",
) -> dict:
set_opening_balance(ledger_path, opening_balance, currency=currency)
imports = [
import_csv(FIXTURE_DIR / "cloud-costs.csv", "cloud", ledger_path=ledger_path),
import_csv(FIXTURE_DIR / "anthropic-billing.csv", "anthropic", ledger_path=ledger_path),
import_csv(FIXTURE_DIR / "hosteurope.csv", "hosteurope", ledger_path=ledger_path),
]
return {
"opening_balance": opening_balance,
"opening_balance_currency": currency,
"imports": [result.as_dict() for result in imports],
}
def build_runway_report(
*,
ledger_path: Path,
opening_balance: float | None = None,
alert_threshold_months: float = 3.0,
currency: str = "EUR",
) -> str:
stats = ledger_stats(ledger_path=ledger_path)
balance = opening_balance if opening_balance is not None else stats["opening_balance"]
if balance is None:
raise ValueError("opening balance required — set via ledger set-balance or --balance")
burns = monthly_burn_series(ledger_path=ledger_path, currency=currency)
runway = compute_runway(
current_balance=balance,
monthly_burns=burns,
alert_threshold_months=alert_threshold_months,
currency=currency,
)
alerts = evaluate_budget_alerts(runway=runway)
generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
report_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
lines = [
f"# Runway Dogfood Evidence — {report_date}",
"",
f"Status: generated v0.1 ({generated_at})",
"",
"Non-secret internal dogfood artefact for RaaS managed-tier runway evidence.",
"Fixture CSVs under `tests/fixtures/` imported into a local SQLite ledger.",
"",
"## Opening position",
"",
f"- Opening balance: **{balance:,.2f} {currency}**",
f"- Alert threshold: **{alert_threshold_months:.1f} months**",
f"- Ledger path: `{ledger_path.resolve()}`",
"",
"## Monthly burn (ledger rollups)",
"",
"| Period | Currency | Total | Entries |",
"| --- | --- | ---: | ---: |",
]
for rollup in monthly_summary(ledger_path=ledger_path):
lines.append(
f"| {rollup.period_month} | {rollup.currency} | {rollup.total:,.2f} | {rollup.entry_count} |"
)
months_display = (
f"{runway.months_remaining:.1f}"
if runway.months_remaining != float("inf")
else "∞"
)
lines.extend(
[
"",
"## Runway projection",
"",
f"- Average monthly burn ({currency}): **{runway.monthly_burn:,.2f}**",
f"- Projected runway: **{months_display} months**",
f"- Below threshold: **{'yes' if runway.below_threshold else 'no'}**",
"",
"## Alerts",
"",
]
)
if alerts:
for alert in alerts:
lines.append(f"- **{alert.code}** ({alert.severity}): {alert.summary}")
else:
lines.append("- No alerts triggered.")
lines.extend(
[
"",
"## Federation signals",
"",
"When `STATE_HUB_API` is reachable, `finhub evaluate --emit` posts:",
"",
"- fin→dev resource pressure (`budget_pressure`, `runway_below_threshold`)",
"- fin→canon viability alert when runway is below threshold",
"",
"## References",
"",
"- `docs/raas-mvp-packaging.md` — managed-tier runway add-on",
"- `workplans/FIN-WP-0001-runway-operations-lane.md` — operational lane",
]
)
return "\n".join(lines) + "\n"
def write_runway_evidence(
*,
ledger_path: Path,
output_dir: Path | None = None,
opening_balance: float | None = None,
seed_fixtures: bool = False,
alert_threshold_months: float = 3.0,
currency: str = "EUR",
) -> Path:
if seed_fixtures:
seed_fixture_ledger(
ledger_path=ledger_path,
opening_balance=opening_balance or 12000.0,
currency=currency,
)
report_dir = output_dir or DEFAULT_EVIDENCE_DIR
report_dir.mkdir(parents=True, exist_ok=True)
report_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
output_path = report_dir / f"runway-dogfood-{report_date}.md"
output_path.write_text(
build_runway_report(
ledger_path=ledger_path,
opening_balance=opening_balance,
alert_threshold_months=alert_threshold_months,
currency=currency,
),
encoding="utf-8",
)
return output_path

303
src/fin_hub/services/ledger.py Normal file
View file

@ -0,0 +1,303 @@
"""SQLite-backed cost ledger for ingested CSV rows."""
from __future__ import annotations
import json
import sqlite3
from dataclasses import asdict, dataclass
from datetime import date, datetime, timezone
from pathlib import Path
from fin_hub.ingest.anthropic import parse_anthropic_billing_csv
from fin_hub.ingest.cloud import parse_cloud_cost_csv
from fin_hub.ingest.hosteurope import parse_hosteurope_csv
DEFAULT_LEDGER_PATH = Path(".fin-hub/ledger.db")
@dataclass(frozen=True)
class LedgerEntry:
source_type: str
category: str
label: str
amount: float
currency: str
period_month: str
incurred_on: date | None
source_path: str
@dataclass(frozen=True)
class MonthlyRollup:
period_month: str
currency: str
total: float
entry_count: int
def as_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class ImportResult:
source_type: str
source_path: str
rows_imported: int
skipped: bool
def as_dict(self) -> dict:
return asdict(self)
def default_ledger_path() -> Path:
return DEFAULT_LEDGER_PATH
def _connect(path: Path) -> sqlite3.Connection:
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
_ensure_schema(conn)
return conn
def _ensure_schema(conn: sqlite3.Connection) -> None:
conn.executescript(
"""
CREATE TABLE IF NOT EXISTS ledger_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
source_type TEXT NOT NULL,
category TEXT NOT NULL,
label TEXT NOT NULL,
amount REAL NOT NULL,
currency TEXT NOT NULL,
period_month TEXT NOT NULL,
incurred_on TEXT,
source_path TEXT NOT NULL,
imported_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS ledger_imports (
source_path TEXT NOT NULL,
source_mtime REAL NOT NULL,
source_type TEXT NOT NULL,
row_count INTEGER NOT NULL,
imported_at TEXT NOT NULL,
PRIMARY KEY (source_path, source_mtime)
);
CREATE TABLE IF NOT EXISTS ledger_meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"""
)
def _utc_now() -> str:
return datetime.now(timezone.utc).isoformat()
def set_opening_balance(path: Path, balance: float, *, currency: str = "EUR") -> None:
with _connect(path) as conn:
conn.execute(
"INSERT OR REPLACE INTO ledger_meta (key, value) VALUES (?, ?)",
("opening_balance", str(balance)),
)
conn.execute(
"INSERT OR REPLACE INTO ledger_meta (key, value) VALUES (?, ?)",
("opening_balance_currency", currency),
)
conn.commit()
def get_opening_balance(path: Path) -> tuple[float | None, str]:
with _connect(path) as conn:
balance_row = conn.execute(
"SELECT value FROM ledger_meta WHERE key = 'opening_balance'"
).fetchone()
currency_row = conn.execute(
"SELECT value FROM ledger_meta WHERE key = 'opening_balance_currency'"
).fetchone()
if balance_row is None:
return None, "EUR"
currency = currency_row["value"] if currency_row else "EUR"
return float(balance_row["value"]), currency
def _entries_from_cloud(path: Path) -> list[LedgerEntry]:
resolved = str(path.resolve())
return [
LedgerEntry(
source_type="cloud",
category=row.service,
label=row.service,
amount=row.amount,
currency=row.currency,
period_month=row.period_month,
incurred_on=row.incurred_on,
source_path=resolved,
)
for row in parse_cloud_cost_csv(path)
]
def _entries_from_anthropic(path: Path) -> list[LedgerEntry]:
resolved = str(path.resolve())
return [
LedgerEntry(
source_type="anthropic",
category=row.provider,
label=row.model,
amount=row.cost,
currency=row.currency,
period_month=row.recorded_at.strftime("%Y-%m"),
incurred_on=row.recorded_at.date(),
source_path=resolved,
)
for row in parse_anthropic_billing_csv(path)
]
def _entries_from_hosteurope(path: Path) -> list[LedgerEntry]:
resolved = str(path.resolve())
return [
LedgerEntry(
source_type="hosteurope",
category=row.service_id,
label=row.title,
amount=row.amount,
currency=row.currency,
period_month=row.period_month,
incurred_on=row.incurred_on,
source_path=resolved,
)
for row in parse_hosteurope_csv(path)
]
_IMPORTERS = {
"cloud": _entries_from_cloud,
"anthropic": _entries_from_anthropic,
"hosteurope": _entries_from_hosteurope,
}
def import_csv(
path: Path,
source_type: str,
*,
ledger_path: Path | None = None,
force: bool = False,
) -> ImportResult:
if source_type not in _IMPORTERS:
raise ValueError(f"Unknown source type '{source_type}'")
ledger = ledger_path or default_ledger_path()
resolved = str(path.resolve())
mtime = path.stat().st_mtime
entries = _IMPORTERS[source_type](path)
imported_at = _utc_now()
with _connect(ledger) as conn:
if not force:
existing = conn.execute(
"SELECT row_count FROM ledger_imports WHERE source_path = ? AND source_mtime = ?",
(resolved, mtime),
).fetchone()
if existing is not None:
return ImportResult(
source_type=source_type,
source_path=resolved,
rows_imported=0,
skipped=True,
)
for entry in entries:
conn.execute(
"""
INSERT INTO ledger_entries (
source_type, category, label, amount, currency,
period_month, incurred_on, source_path, imported_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
entry.source_type,
entry.category,
entry.label,
entry.amount,
entry.currency,
entry.period_month,
entry.incurred_on.isoformat() if entry.incurred_on else None,
entry.source_path,
imported_at,
),
)
conn.execute(
"""
INSERT OR REPLACE INTO ledger_imports (
source_path, source_mtime, source_type, row_count, imported_at
) VALUES (?, ?, ?, ?, ?)
""",
(resolved, mtime, source_type, len(entries), imported_at),
)
conn.commit()
return ImportResult(
source_type=source_type,
source_path=resolved,
rows_imported=len(entries),
skipped=False,
)
def monthly_summary(*, ledger_path: Path | None = None) -> list[MonthlyRollup]:
ledger = ledger_path or default_ledger_path()
with _connect(ledger) as conn:
rows = conn.execute(
"""
SELECT period_month, currency, SUM(amount) AS total, COUNT(*) AS entry_count
FROM ledger_entries
GROUP BY period_month, currency
ORDER BY period_month, currency
"""
).fetchall()
return [
MonthlyRollup(
period_month=row["period_month"],
currency=row["currency"],
total=float(row["total"]),
entry_count=int(row["entry_count"]),
)
for row in rows
]
def monthly_burn_series(
*,
ledger_path: Path | None = None,
currency: str = "EUR",
) -> list[float]:
rollups = monthly_summary(ledger_path=ledger_path)
burns = [rollup.total for rollup in rollups if rollup.currency == currency]
return burns
def ledger_stats(*, ledger_path: Path | None = None) -> dict:
ledger = ledger_path or default_ledger_path()
balance, balance_currency = get_opening_balance(ledger)
rollups = monthly_summary(ledger_path=ledger)
with _connect(ledger) as conn:
entry_count = conn.execute("SELECT COUNT(*) AS c FROM ledger_entries").fetchone()["c"]
import_count = conn.execute("SELECT COUNT(*) AS c FROM ledger_imports").fetchone()["c"]
return {
"ledger_path": str(ledger.resolve()),
"opening_balance": balance,
"opening_balance_currency": balance_currency,
"entry_count": entry_count,
"import_count": import_count,
"monthly_rollups": [rollup.as_dict() for rollup in rollups],
}
def ledger_stats_json(*, ledger_path: Path | None = None) -> str:
return json.dumps(ledger_stats(ledger_path=ledger_path), indent=2)

60
tests/test_ledger.py Normal file
View file

@ -0,0 +1,60 @@
from pathlib import Path
import pytest
from fin_hub.services.evaluate import evaluate_runway
from fin_hub.services.evidence import build_runway_report, seed_fixture_ledger, write_runway_evidence
from fin_hub.services.ledger import import_csv, monthly_burn_series, monthly_summary, set_opening_balance
FIXTURES = Path(__file__).parent / "fixtures"
def test_ledger_import_and_monthly_summary(tmp_path: Path):
ledger = tmp_path / "ledger.db"
set_opening_balance(ledger, 12000.0)
cloud = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
host = import_csv(FIXTURES / "hosteurope.csv", "hosteurope", ledger_path=ledger)
assert cloud.rows_imported == 3
assert host.rows_imported == 2
rollups = monthly_summary(ledger_path=ledger)
eur_totals = {rollup.period_month: rollup.total for rollup in rollups if rollup.currency == "EUR"}
assert eur_totals["2026-06"] == pytest.approx(85.5 + 99.8)
burns = monthly_burn_series(ledger_path=ledger, currency="EUR")
assert burns == [pytest.approx(185.3)]
def test_ledger_import_skips_unchanged_file(tmp_path: Path):
ledger = tmp_path / "ledger.db"
first = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
second = import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
assert first.rows_imported == 3
assert second.skipped is True
assert second.rows_imported == 0
def test_evaluate_runway_from_ledger(tmp_path: Path):
ledger = tmp_path / "ledger.db"
seed_fixture_ledger(ledger_path=ledger, opening_balance=12000.0)
report = evaluate_runway(ledger_path=ledger, alert_threshold_months=3.0)
assert report["runway"]["current_balance"] == 12000.0
assert report["runway"]["monthly_burn"] > 0
def test_build_runway_report_and_write_evidence(tmp_path: Path):
ledger = tmp_path / "ledger.db"
seed_fixture_ledger(ledger_path=ledger, opening_balance=12000.0)
text = build_runway_report(ledger_path=ledger)
assert "Runway Dogfood Evidence" in text
assert "2026-06" in text
output = write_runway_evidence(
ledger_path=ledger,
output_dir=tmp_path / "evidence",
opening_balance=12000.0,
)
assert output.exists()
assert "runway-dogfood-" in output.name

6
workplans/FIN-WP-0001-runway-operations-lane.md
View file

@ -54,7 +54,7 @@ spine (resolves C-24).
```task
id: FIN-WP-0001-T03
status: todo
status: done
priority: high
state_hub_task_id: "9350d324-6194-44d5-9923-3547d69a2eed"
```
@ -66,7 +66,7 @@ CLI subcommands: `ledger import`, `ledger summary`.
```task
id: FIN-WP-0001-T04
status: todo
status: done
priority: medium
state_hub_task_id: "9574806e-e414-421d-8b58-844ccc9e6af7"
```
@ -79,7 +79,7 @@ cron or systemd timer.
```task
id: FIN-WP-0001-T05
status: todo
status: done
priority: medium
state_hub_task_id: "0b3b5e95-8c72-4e70-9171-14ef23c9e969"
```