Compare commits

...

10 Commits

  1. 9
      README.md
  2. 6
      config/urls.py
  3. 1
      core/context.py
  4. 10
      core/forms.py
  5. 4
      core/templates/core/dashboard.html
  6. 9
      core/templates/core/reconciliations/history.html
  7. 31
      core/templates/core/reconciliations/index.html
  8. 110
      core/views.py
  9. 2
      docs/imports.md
  10. 18
      docs/local-workflows.md
  11. 291
      docs/superpowers/plans/2026-09-08-reconciliation-checkpoints.md
  12. 99
      docs/superpowers/specs/2026-09-08-reconciliation-design.md
  13. 31
      finance/migrations/0003_reconciliationcheckpoint.py
  14. 3
      finance/models/__init__.py
  15. 12
      finance/models/reconciliation.py
  16. 96
      finance/reconciliation.py
  17. 3
      finance/services/__init__.py
  18. 25
      finance/services/reconciliation.py
  19. 6
      ingestion/parsers.py
  20. 346
      tests/finance/test_reconciliation.py
  21. 12
      tests/test_import_parsers.py
  22. 60
      tests/test_mvp_acceptance.py
  23. 320
      tests/test_reconciliation_views.py
  24. 37
      tests/test_report_views.py

9
README.md

@ -31,6 +31,15 @@ Start with [contributor instructions](AGENTS.md), then read the
Owners can follow the copy-paste [local workflow guide](docs/local-workflows.md)
from reviewed import through reports, export, and verified backup restoration.
## Reconciliation
Open **Reconciliation**, select an account, and enter a checkpoint date with the
institution balance. Review the preview before confirming the checkpoint.
Checkpoint dates are inclusive, so they include activity posted on that date. A
new checkpoint corrects a prior checkpoint. Newly imported activity or a later
backdated transaction can change a previously reconciled status, so review the
status and record a new checkpoint after verifying the institution balance.
## Architecture
The approved technical choices are:

6
config/urls.py

@ -7,6 +7,12 @@ urlpatterns = [
path("", views.home, name="home"),
path("health/", views.health, name="health"),
path("reports/monthly/", views.monthly_report, name="monthly-report"),
path("reconciliations/", views.reconciliation_index, name="reconciliation-index"),
path(
"reconciliations/<uuid:account_id>/",
views.reconciliation_history,
name="reconciliation-history",
),
path("exports/", views.export_index, name="export-index"),
path("exports/download/", views.export_download, name="export-download"),
path("transactions/", views.transaction_list, name="transaction-list"),

1
core/context.py

@ -5,6 +5,7 @@ def navigation(_request):
("transaction-list", "Transactions"),
("category-list", "Categories"),
("payee-list", "Payees"),
("reconciliation-index", "Reconciliation"),
("monthly-report", "Reports"),
("export-index", "Exports"),
)}

10
core/forms.py

@ -91,6 +91,16 @@ class PayeeForm(forms.Form):
name = forms.CharField(max_length=255)
class ReconciliationForm(forms.Form):
account = forms.TypedChoiceField(coerce=UUID)
as_of_date = forms.DateField(widget=forms.DateInput(attrs={"type": "date"}))
institution_balance = forms.DecimalField(max_digits=18, decimal_places=2)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.fields["account"].choices = _reference_choices(Account)
class ReportSelectionForm(forms.Form):
month = forms.CharField(
error_messages={"required": "invalid_month"},

4
core/templates/core/dashboard.html

@ -25,5 +25,9 @@
{% if balances %}<div class="table-wrap"><table><thead><tr><th>Account</th><th class="number">Balance</th></tr></thead><tbody>
{% for balance in balances %}<tr><td>{{ balance.account_name }}</td><td class="number">{% if balance.balance < 0 %}<span>Balance (negative): </span>{% endif %}{{ balance.balance|floatformat:"2g" }} {{ summary.currency }}</td></tr>{% endfor %}
</tbody></table></div>{% else %}<p>No accounts use {{ summary.currency }}.</p>{% endif %}
<h2>Reconciliation status</h2>
{% if reconciliation_statuses %}<div class="table-wrap"><table><thead><tr><th>Account</th><th>As of</th><th>Status</th><th class="number">Live difference</th></tr></thead><tbody>
{% for row in reconciliation_statuses %}<tr><td><a href="{% url 'reconciliation-history' row.account_id %}">{{ row.account_name }}</a></td><td>{{ row.as_of_date|date:"Y-m-d" }}</td><td>{% if row.state == "reconciled" %}Reconciled{% elif row.state == "changed_since_reconciliation" %}Changed since reconciliation{% else %}Difference{% endif %}</td><td class="number">{{ row.live_difference|floatformat:"2g" }} {{ row.currency }}</td></tr>{% endfor %}
</tbody></table></div>{% else %}<p>No reconciliation checkpoints for active BRL accounts.</p>{% endif %}
{% endif %}
{% endblock %}

9
core/templates/core/reconciliations/history.html

@ -0,0 +1,9 @@
{% extends "core/base.html" %}
{% block title %}{{ account.name }} reconciliation · Gizmoduck{% endblock %}
{% block content %}
<div class="heading"><h1>{{ account.name }} reconciliation history</h1><a class="button secondary" href="{% url 'reconciliation-index' %}">New checkpoint</a></div>
{% if account.archived %}<p>Archived account</p>{% endif %}
{% if history %}<div class="table-wrap"><table><thead><tr><th>As of</th><th>Created at</th><th class="number">Institution balance</th><th class="number">Snapshot ledger balance</th><th class="number">Snapshot difference</th><th class="number">Live ledger balance</th><th class="number">Live difference</th><th>Status</th></tr></thead><tbody>
{% for row in history %}<tr><td>{{ row.as_of_date|date:"Y-m-d" }}</td><td>{{ row.created_at|date:"Y-m-d H:i:s T" }}</td><td class="number">{{ row.institution_balance|floatformat:"2g" }} {{ row.currency }}</td><td class="number">{{ row.snapshot_ledger_balance|floatformat:"2g" }} {{ row.currency }}</td><td class="number">{{ row.snapshot_difference|floatformat:"2g" }} {{ row.currency }}</td><td class="number">{{ row.live_ledger_balance|floatformat:"2g" }} {{ row.currency }}</td><td class="number">{{ row.live_difference|floatformat:"2g" }} {{ row.currency }}</td><td>{% if row.state == "reconciled" %}Reconciled{% elif row.state == "changed_since_reconciliation" %}Changed since reconciliation{% else %}Difference{% endif %}</td></tr>{% endfor %}
</tbody></table></div>{% else %}<p>No reconciliation checkpoints for this account.</p>{% endif %}
{% endblock %}

31
core/templates/core/reconciliations/index.html

@ -0,0 +1,31 @@
{% extends "core/base.html" %}
{% block title %}Reconciliation · Gizmoduck{% endblock %}
{% block content %}
<h1>Reconciliation</h1>
<p>Compare an institution balance with the ledger through an inclusive date.</p>
<form method="post">
{% csrf_token %}
{{ form.non_field_errors }}
<p><label for="id_account">Account</label>{{ form.account }}{% for error in form.account.errors %}<span class="error">{{ error }}</span>{% endfor %}</p>
<p><label for="id_as_of_date">As-of date</label>{{ form.as_of_date }}{% for error in form.as_of_date.errors %}<span class="error">{{ error }}</span>{% endfor %}</p>
<p><label for="id_institution_balance">Institution balance</label>{{ form.institution_balance }}{% for error in form.institution_balance.errors %}<span class="error">{{ error }}</span>{% endfor %}</p>
<button type="submit" name="action" value="preview">Preview</button>
{% if preview %}
<input type="hidden" name="preview_token" value="{{ preview_token }}">
<h2>Preview</h2>
<dl class="details">
<dt>Account</dt><dd>{{ preview.account.name }}</dd>
<dt>As-of date</dt><dd>{{ preview.as_of_date|date:"Y-m-d" }}</dd>
<dt>Institution balance</dt><dd class="number">{{ preview.institution_balance|floatformat:"2g" }} {{ preview.account.currency }}</dd>
<dt>Ledger balance</dt><dd class="number">{{ preview.ledger_balance|floatformat:"2g" }} {{ preview.account.currency }}</dd>
<dt>Difference</dt><dd class="number">{{ preview.difference|floatformat:"2g" }} {{ preview.account.currency }}</dd>
</dl>
<p>If you change the account, date, or institution balance, choose Preview again before confirming.</p>
<button type="submit" name="action" value="confirm">Confirm checkpoint</button>
{% endif %}
</form>
<h2>Latest reconciliation status</h2>
{% if latest_statuses %}<div class="table-wrap"><table><thead><tr><th>Account</th><th>As of</th><th>Status</th><th class="number">Live difference</th></tr></thead><tbody>
{% for row in latest_statuses %}<tr><td><a href="{% url 'reconciliation-history' row.account_id %}">{{ row.account_name }}</a></td><td>{{ row.as_of_date|date:"Y-m-d" }}</td><td>{% if row.state == "reconciled" %}Reconciled{% elif row.state == "changed_since_reconciliation" %}Changed since reconciliation{% else %}Difference{% endif %}</td><td class="number">{{ row.live_difference|floatformat:"2g" }} {{ row.currency }}</td></tr>{% endfor %}
</tbody></table></div>{% else %}<p>No reconciliation checkpoints for active accounts.</p>{% endif %}
{% endblock %}

110
core/views.py

@ -1,4 +1,9 @@
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from django.contrib import messages
from django.core import signing
from django.core.paginator import Paginator
from django.db import DatabaseError, connection
from django.http import HttpResponse, JsonResponse
@ -12,15 +17,21 @@ from finance.queries import (
QueryInputError, TransactionFilters, filtered_transactions,
listed_accounts, listed_categories, listed_payees,
)
from finance.reconciliation import (
checkpoint_history,
latest_checkpoint_statuses,
ledger_balance_through,
)
from finance.reports import account_balances, available_currencies, monthly_summary, spending_by_category
from finance.services import (
Actor, DomainError, annotate_transaction, archive, create_account,
create_category, create_payee, record_transaction,
create_category, create_payee, create_reconciliation_checkpoint,
record_transaction,
)
from .forms import (
AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm,
ReportSelectionForm,
ReconciliationForm, ReportSelectionForm,
)
@ -38,6 +49,18 @@ DOMAIN_MESSAGES = {
UNKNOWN_DOMAIN_MESSAGE = "Unable to save this change."
@dataclass(frozen=True)
class ReconciliationPreview:
account: Account
as_of_date: date
institution_balance: Decimal
ledger_balance: Decimal
@property
def difference(self) -> Decimal:
return self.institution_balance - self.ledger_balance
def health(request):
try:
with connection.cursor() as cursor:
@ -56,6 +79,73 @@ def monthly_report(request):
return _monthly_report(request, "core/reports/monthly.html")
def reconciliation_index(request):
form = ReconciliationForm(request.POST or None)
preview = None
preview_token = None
if request.method == "POST" and form.is_valid():
action = request.POST.get("action")
values = form.cleaned_data
preview_inputs = {
"account": str(values["account"]),
"as_of_date": values["as_of_date"].isoformat(),
"institution_balance": format(values["institution_balance"], ".2f"),
}
if action == "preview":
selected_account = get_object_or_404(
Account,
pk=values["account"],
archived=False,
)
preview = ReconciliationPreview(
account=selected_account,
as_of_date=values["as_of_date"],
institution_balance=values["institution_balance"],
ledger_balance=ledger_balance_through(
selected_account.pk,
values["as_of_date"],
),
)
preview_token = signing.dumps(preview_inputs, salt="reconciliation-preview")
elif action == "confirm":
try:
confirmed_inputs = signing.loads(
request.POST.get("preview_token", ""),
salt="reconciliation-preview",
)
except signing.BadSignature:
confirmed_inputs = None
if confirmed_inputs != preview_inputs:
form.add_error(None, "Preview these values before confirming the checkpoint.")
else:
try:
saved = create_reconciliation_checkpoint(
account_id=values["account"],
as_of_date=values["as_of_date"],
institution_balance=values["institution_balance"],
actor=LOCAL_ACTOR,
)
except DomainError as exc:
form.add_error(None, _domain_message(exc))
else:
messages.success(request, "Reconciliation checkpoint recorded.")
return redirect("reconciliation-history", account_id=saved.account_id)
return render(request, "core/reconciliations/index.html", {
"form": form,
"preview": preview,
"preview_token": preview_token,
"latest_statuses": _latest_reconciliation_statuses(),
})
def reconciliation_history(request, account_id):
selected_account = get_object_or_404(Account, pk=account_id)
return render(request, "core/reconciliations/history.html", {
"account": selected_account,
"history": checkpoint_history(selected_account.pk),
})
def export_index(request):
return render(request, "core/exports.html")
@ -236,6 +326,8 @@ def _monthly_report(request, template_name):
}
form = ReportSelectionForm(parameters, currencies=currencies)
context = {"selection_form": form}
if template_name == "core/dashboard.html":
context["reconciliation_statuses"] = latest_checkpoint_statuses("BRL")
status = 200
if form.is_valid():
month = form.cleaned_data["month"]
@ -248,3 +340,17 @@ def _monthly_report(request, template_name):
else:
status = 400
return render(request, template_name, context, status=status)
def _latest_reconciliation_statuses():
currencies = (
Account.objects.filter(archived=False)
.order_by("currency")
.values_list("currency", flat=True)
.distinct()
)
return [
status
for currency in currencies
for status in latest_checkpoint_statuses(currency)
]

2
docs/imports.md

@ -138,7 +138,7 @@ both occurrence and posting dates. OFX and PDF are unsupported.
| Profile | Accepted layout |
| --- | --- |
| `BANK_CSV_V1` | Bank account CSV, comma-delimited with CSV quoting, at least six header columns; date at column 0 as `DD/MM/YYYY`, description at 2, signed decimal-point amount at 5. Each record must match the header width. |
| `OUROCARD_FULL_V1` | Card TXT, `DD.MM.YYYY`, 37-character description/location field, two-character country field, debit and credit in Brazilian decimal notation. A separate `PGTO DEBITO CONTA` layout is recognized. |
| `OUROCARD_FULL_V1` | Card TXT, `DD.MM.YYYY`, with either a 37-character description/location field plus a two-character country field, or a 33-character description/location field plus a seven-character country field; debit and credit use Brazilian decimal notation. A separate `PGTO DEBITO CONTA` layout is recognized. |
| `OUROCARD_PARTIAL_V1` | Card TXT, `DD/MM`, 43-character description/location field, country, debit and credit; requires an explicit date interval. |
CSV header aliases are case/accent insensitive: date accepts `Data`,

18
docs/local-workflows.md

@ -25,8 +25,8 @@ local database tools such as DBeaver. The database name is `gizmoduck`. Use
routine inspection. The migration and bootstrap roles have broader privileges
and are for maintenance only.
The navigation provides Accounts, Transactions, Categories, Payees, Reports,
and Exports. Create accounts, categories, and payees before assigning them.
The navigation provides Accounts, Transactions, Categories, Payees, Reconciliation,
Reports, and Exports. Create accounts, categories, and payees before assigning them.
References may be archived, but historical transactions continue to show them.
Transaction editing is limited to description, payee, and category. Amount,
dates, account, kind, import provenance, transfers, refunds, statements, and
@ -104,6 +104,20 @@ excluded from cash flow. A card purchase is counted once as an expense. Spending
is grouped by expense category with a visible Uncategorized row, and account
balances remain separate by account and currency.
## Reconcile an account balance
Open **Reconciliation** from the navigation. Select the account, enter the
checkpoint date and the institution balance, then choose **Preview** to review
the ledger balance and difference. Choose **Confirm checkpoint** only after the
preview is correct. Checkpoint dates are inclusive, so the ledger includes every
transaction posted on that date.
A checkpoint is immutable. To correct an earlier checkpoint, record a new one
for the appropriate account and date. Newly imported activity or a transaction
added later with a backdated posting date can change a previously reconciled
status; review the reconciliation status and record a new checkpoint when the
institution balance has been verified.
## Export portable data
Use `http://127.0.0.1:8000/exports/` to download a ZIP, or create one under the

291
docs/superpowers/plans/2026-09-08-reconciliation-checkpoints.md

@ -0,0 +1,291 @@
# Reconciliation Checkpoints Implementation Plan
> **For Codex:** Use the `superpowers:executing-plans` skill to carry out this plan task by task.
**Goal:** Let the owner record immutable institution-balance checkpoints and see whether Gizmoduck's ledger still agrees with them.
**Architecture:** A reconciliation checkpoint is an audited financial write that snapshots an account's ledger total through an inclusive date. Read-model helpers recalculate the live ledger total, derive a status, and supply the reconciliation page and dashboard. A preview-and-confirm UI ensures the owner sees the calculated difference before the checkpoint is written.
**Tech Stack:** Python 3.12, Django, PostgreSQL, Django migrations, pytest, Docker Compose, server-rendered Django templates.
**Scope constraints:** Preserve all existing financial data. A checkpoint never changes transactions, imports, opening balances, or account activation. It is immutable; a correction creates a new row. Use only synthetic test data and never commit data from `imports/`, databases, backups, logs, or screenshots.
---
## Task 1: Persist audited checkpoint snapshots
**Files:**
- Create: `finance/models/reconciliation.py`
- Modify: `finance/models/__init__.py`
- Create: `finance/migrations/0003_reconciliationcheckpoint.py`
- Create: `finance/services/reconciliation.py`
- Modify: `finance/services/__init__.py`
- Create: `tests/finance/test_reconciliation.py`
- [ ] **Step 1: Write failing domain-service tests with synthetic records.**
Cover an inclusive `as_of_date`, a zero institution balance, two corrections for the same account/date, and audit metadata. The essential behaviour is:
```python
checkpoint = create_reconciliation_checkpoint(
account_id=account.id,
as_of_date=date(2026, 9, 1),
institution_balance=Decimal("0.00"),
actor=Actor("HUMAN", "local-owner"),
)
assert checkpoint.ledger_balance_at_creation == Decimal("-42.50")
assert checkpoint.difference_at_creation == Decimal("42.50")
assert AuditEvent.objects.filter(
action="reconciliation_checkpoint.created",
object_id=checkpoint.id,
).exists()
```
Run: `docker compose run --rm test pytest -q tests/finance/test_reconciliation.py`
Expected: FAIL because the model and service do not exist.
- [ ] **Step 2: Add the checkpoint model and migration.**
In `finance/models/reconciliation.py`, define `ReconciliationCheckpoint(TimestampedModel)` with:
```python
account = models.ForeignKey(Account, on_delete=models.PROTECT)
as_of_date = models.DateField()
institution_balance = models.DecimalField(max_digits=18, decimal_places=2)
ledger_balance_at_creation = models.DecimalField(max_digits=18, decimal_places=2)
difference_at_creation = models.DecimalField(max_digits=18, decimal_places=2)
```
Export it from `finance/models/__init__.py`. Generate and inspect a migration named `0003_reconciliationcheckpoint.py`; it must not introduce a uniqueness constraint for account/date or a delete/cascade path that removes history with the account.
- [ ] **Step 3: Implement one transactional, audited write service.**
`create_reconciliation_checkpoint` must be decorated with the repository's `@write` mechanism, lock the active account using the existing `lock(Account, ..., active=True)` helper, calculate a decimal ledger total from `LedgerEntry` rows where `posting_date__lte=as_of_date`, and persist the snapshot values.
```python
@write(action="reconciliation_checkpoint.created")
def create_reconciliation_checkpoint(*, account_id, as_of_date, institution_balance, actor):
account = lock(Account, account_id, active=True)
ledger_balance = ledger_balance_through(account.id, as_of_date)
return ReconciliationCheckpoint.objects.create(
account=account,
as_of_date=as_of_date,
institution_balance=money(institution_balance, allow_zero=True),
ledger_balance_at_creation=ledger_balance,
difference_at_creation=money(institution_balance - ledger_balance, allow_zero=True),
)
```
Keep the aggregation private to the service at this point; Task 2 will extract the shared read helper. Add a test proving an archived account raises the existing domain error and does not create either a checkpoint or audit event.
- [ ] **Step 4: Verify and commit the cohesive domain slice.**
Run:
```powershell
docker compose run --rm test pytest -q tests/finance/test_reconciliation.py
docker compose run --rm web python manage.py makemigrations --check --dry-run
docker compose run --rm web python manage.py check
```
Stage only the seven Task 1 files. Inspect staged names and full staged content for real financial data or secrets, then commit with repository-local identity:
```powershell
git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: record reconciliation checkpoints"
```
## Task 2: Add live reconciliation status read models
**Files:**
- Create: `finance/reconciliation.py`
- Modify: `finance/services/reconciliation.py`
- Modify: `tests/finance/test_reconciliation.py`
- [ ] **Step 1: Add failing status tests.**
Add synthetic tests for these derived states:
| Snapshot difference | Later ledger activity through date | Expected state |
| --- | --- | --- |
| `0.00` | none | `reconciled` |
| non-zero | none | `difference` |
| `0.00` | backdated activity | `changed_since_reconciliation` |
Also test that, when several checkpoints exist for one account/date, the newest creation is retained in history and the latest status selection uses the latest `as_of_date`, then creation time as its tie-breaker.
Run: `docker compose run --rm test pytest -q tests/finance/test_reconciliation.py`
Expected: FAIL because no read model exists.
- [ ] **Step 2: Extract the inclusive ledger-total helper.**
Put the shared query in `finance/reconciliation.py`. It returns `Decimal("0.00")` when there are no entries and intentionally includes opening entries like every other ledger entry.
```python
def ledger_balance_through(account_id: UUID, as_of_date: date) -> Decimal:
return money(
LedgerEntry.objects.filter(
account_id=account_id,
posting_date__lte=as_of_date,
).aggregate(total=Coalesce(Sum("amount"), Decimal("0.00")))["total"],
allow_zero=True,
)
```
Update the write service to use this helper, preserving its transaction and audit behaviour.
- [ ] **Step 3: Implement presentation-neutral status values.**
Define an immutable `CheckpointStatus` dataclass containing checkpoint/account identifiers, account name, currency, as-of date, institution balance, snapshot ledger/difference, live ledger/difference, and state. Implement:
```python
def checkpoint_status(checkpoint: ReconciliationCheckpoint) -> CheckpointStatus: ...
def checkpoint_history(account_id: UUID) -> list[CheckpointStatus]: ...
def latest_checkpoint_statuses(currency: str) -> list[CheckpointStatus]: ...
```
Compute `live_difference = institution_balance - live_ledger_balance`. State selection is exactly:
```python
if live_difference == Decimal("0.00"):
state = "reconciled"
elif checkpoint.difference_at_creation == Decimal("0.00"):
state = "changed_since_reconciliation"
else:
state = "difference"
```
`latest_checkpoint_statuses` selects one checkpoint per active account in the requested currency, ordered by `as_of_date DESC`, `created_at DESC`, `id DESC`. It does not alter the checkpoint history.
- [ ] **Step 4: Verify and commit the read-model slice.**
Run: `docker compose run --rm test pytest -q tests/finance/test_reconciliation.py`
Stage only Task 2 files, inspect staged content for sensitive data, and commit:
```powershell
git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: report reconciliation status"
```
## Task 3: Build the reconciliation pages and dashboard summary
**Files:**
- Modify: `config/urls.py`
- Modify: `core/context.py`
- Modify: `core/forms.py`
- Modify: `core/views.py`
- Modify: `core/templates/core/base.html`
- Modify: `core/templates/core/dashboard.html`
- Create: `core/templates/core/reconciliations/index.html`
- Create: `core/templates/core/reconciliations/history.html`
- Create: `tests/test_reconciliation_views.py`
- Modify: `tests/test_report_views.py`
- [ ] **Step 1: Write failing end-to-end view tests.**
Use synthetic account and ledger data. Cover:
- the navigation link and reconciliation landing page;
- active accounts only in the form choices;
- a `preview` submission renders ledger balance and difference but creates no checkpoint;
- a `confirm` submission revalidates the values, creates one checkpoint through the service, then redirects to that account's history;
- malformed money/date values show field errors and no write;
- CSRF protection rejects a client without the token;
- archived-account confirmation fails safely with no write;
- history displays snapshot and live values for corrected same-date checkpoints; and
- dashboard displays the current status after later backdated activity.
Run: `docker compose run --rm test pytest -q tests/test_reconciliation_views.py tests/test_report_views.py`
Expected: FAIL because routes, forms, views, and templates do not exist.
- [ ] **Step 2: Wire navigation, routes, and form validation.**
Add a Reconciliation navigation item in `core/context.py`, routes named `reconciliation-index` and `reconciliation-history` in `config/urls.py`, and `ReconciliationForm` in `core/forms.py` with active `Account` choices, date, and money amount. Reuse the repository's reference-choice helpers and decimal validation; zero must remain valid.
- [ ] **Step 3: Implement a preview-and-confirm write flow.**
The index view receives two explicit form actions:
```python
if action == "preview" and form.is_valid():
preview = ReconciliationPreview(
account=form.cleaned_data["account"],
as_of_date=form.cleaned_data["as_of_date"],
institution_balance=form.cleaned_data["institution_balance"],
ledger_balance=ledger_balance_through(...),
)
elif action == "confirm" and form.is_valid():
checkpoint = create_reconciliation_checkpoint(..., actor=LOCAL_OWNER)
return redirect("reconciliation-history", account_id=checkpoint.account_id)
```
Render a confirmation submit only after a valid preview. Confirmation must call the service again rather than trusting hidden snapshot fields, so changes between preview and confirmation receive a fresh ledger calculation. The history view can display an archived account's existing checkpoints but must never offer it as a new checkpoint choice.
- [ ] **Step 4: Render readable status and history.**
The index template presents the form, preview, latest status per active account, and clear labels for Reconciled, Difference, and Changed since reconciliation. The history template presents every checkpoint in descending as-of/creation order, with institution, snapshot ledger/difference, live ledger/difference, and status. Use plain text and existing currency formatting; do not expose source-file contents or parser errors.
Add `latest_checkpoint_statuses("BRL")` to dashboard context and render a compact reconciliation-status table beneath account balances. Preserve existing dashboard and monthly-report behaviour.
- [ ] **Step 5: Verify and commit the web slice.**
Run:
```powershell
docker compose run --rm test pytest -q tests/test_reconciliation_views.py tests/test_report_views.py tests/finance/test_reconciliation.py
docker compose run --rm web python manage.py check
```
Stage only Task 3 files, inspect staged paths and content for sensitive data, then commit:
```powershell
git -c user.name=Codex -c user.email=codex@localhost commit -m "feat: add reconciliation workflow"
```
## Task 4: Document the owner workflow and run acceptance verification
**Files:**
- Modify: `docs/local-workflows.md`
- Modify: `README.md`
- Modify: `tests/test_mvp_acceptance.py`
- [ ] **Step 1: Add a synthetic acceptance test for the owner path.**
Extend the acceptance suite with a user-level flow: create a synthetic account and transaction, preview and confirm a matching checkpoint, then add a synthetic backdated transaction and assert the dashboard reports “Changed since reconciliation.” This validates the cross-layer outcome without importing or inspecting any real local data.
Run: `docker compose run --rm test pytest -q tests/test_mvp_acceptance.py`
Expected: FAIL until the completed slices are integrated.
- [ ] **Step 2: Document recording and correcting a checkpoint.**
In `docs/local-workflows.md` and `README.md`, add concise owner instructions: open Reconciliation, enter an account/date/institution balance, review the preview, confirm, and use a new checkpoint to correct a prior one. Explain that checkpoint dates are inclusive and that newly imported or backdated transactions can change a previously reconciled status. Do not include real account names, balances, paths, or import details.
- [ ] **Step 3: Run final verification.**
Run:
```powershell
docker compose build
docker compose run --rm test pytest -q
docker compose run --rm web python manage.py migrate --check
docker compose run --rm web python manage.py check
docker compose run --rm web python manage.py test_permission_matrix
docker compose up -d web
Invoke-RestMethod http://127.0.0.1:8000/health/ | ConvertTo-Json -Compress
git diff --check
git status --short
```
Confirm the test run has no unexpected failures, migration check is clean, permission matrix passes, health reports `ok`, and the working tree contains only the intended documentation and test changes.
- [ ] **Step 4: Commit the documentation and acceptance coverage.**
Stage only the Task 4 files. Inspect staged names and full content for data-policy compliance, then commit:
```powershell
git -c user.name=Codex -c user.email=codex@localhost commit -m "docs: explain reconciliation workflow"
```

99
docs/superpowers/specs/2026-09-08-reconciliation-design.md

@ -0,0 +1,99 @@
# Reconciliation checkpoints design
## Purpose
Add a local reconciliation workflow that compares an institution-reported
balance with Gizmoduck's ledger for one account on one date. It gives the owner
auditable confidence in account balances without changing transactions,
creating clearing state, or depending on a particular import format.
## Scope
The first version supports every active bank or credit-card account and any
chosen as-of date. It records immutable balance checkpoints, shows their
current status, and preserves their original calculation for audit.
It does not clear individual transactions, lock a reconciled period, infer a
balance from source files, create card statements, or modify financial rows.
Those capabilities remain future enhancements.
## Domain model
Add `ReconciliationCheckpoint` to the finance domain:
| Field | Meaning |
| --- | --- |
| `account` | Active account being checked; protected from deletion. |
| `as_of_date` | Inclusive posting-date boundary for the comparison. |
| `institution_balance` | Balance shown by the bank or card institution, in the account currency. Zero is valid. |
| `ledger_balance_at_creation` | Sum of account transaction amounts through `as_of_date` when the checkpoint was recorded. |
| `difference_at_creation` | `institution_balance - ledger_balance_at_creation`. |
| timestamps | Immutable creation and update timestamps inherited from the existing base model. |
There is deliberately no uniqueness constraint on `(account, as_of_date)`.
Entering a corrected balance creates another checkpoint rather than rewriting
or deleting history. The latest checkpoint by creation time is the active
checkpoint for that account and date.
The reconciliation service owns creation. It validates the account is active,
accepts decimal-safe balances including zero, calculates the ledger snapshot in
one database transaction, saves the checkpoint, and writes the existing
`AuditEvent` with the human actor and outcome. No imported source text or
institution credentials are persisted.
## Current status and drift
When displaying a checkpoint, Gizmoduck recalculates the account's live ledger
balance through the inclusive `as_of_date`. It compares that value with the
stored institution balance.
* **Reconciled**: the live difference is zero.
* **Difference**: the live difference is nonzero and the original checkpoint
also did not reconcile.
* **Changed since reconciliation**: the original checkpoint reconciled, but
later backdated activity changed the live ledger balance.
The checkpoint continues to retain its original ledger balance and difference.
This makes late imports and dated manual entries visible without mutating the
historical record. Opening balances participate in the ledger sum exactly like
other transactions; reports still exclude them from income and spending.
## Owner workflow
Add a Reconciliation navigation entry and page.
The page lists active accounts with their most recent checkpoint date and live
status. The owner can select an account, choose an as-of date, and enter the
institution balance. Before submission, the form shows the current ledger
balance and difference for confirmation. Saving records the immutable
checkpoint and redirects to the account's reconciliation history.
History shows each checkpoint's as-of date, institution balance, original
ledger balance, original difference, current ledger balance, current
difference, created time, and status. Amounts never cross currency boundaries.
The existing dashboard retains account balances and gains only a compact latest
reconciliation status per account.
## Error handling and security
Invalid dates and money values receive field-level form errors. Archived or
missing accounts are rejected by the domain service and translated to the
existing safe owner-facing error messages. Forms use the existing CSRF
protection and fixed local `HUMAN/local-owner` actor. Database exceptions and
financial source data are never exposed in responses.
## Verification
Use only synthetic data. Tests cover:
* exact bank and credit-card reconciliation, including a zero institution
balance;
* an initial difference and a later corrected checkpoint for the same date;
* a backdated transaction changing a previously reconciled checkpoint;
* opening balances in reconciliation totals;
* archived account rejection and audit behavior;
* migration constraints, owner form validation, CSRF protection, account
history, and dashboard status rendering.
Run focused domain and web tests, the full pytest suite, Django migration and
system checks, and the existing PostgreSQL permission check.

31
finance/migrations/0003_reconciliationcheckpoint.py

@ -0,0 +1,31 @@
# Generated by Django 5.2.17 on 2026-09-09 00:13
import django.db.models.deletion
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('finance', '0002_transaction_report_indexes'),
]
operations = [
migrations.CreateModel(
name='ReconciliationCheckpoint',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
('as_of_date', models.DateField()),
('institution_balance', models.DecimalField(decimal_places=2, max_digits=18)),
('ledger_balance_at_creation', models.DecimalField(decimal_places=2, max_digits=18)),
('difference_at_creation', models.DecimalField(decimal_places=2, max_digits=18)),
('account', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='finance.account')),
],
options={
'abstract': False,
},
),
]

3
finance/models/__init__.py

@ -1,8 +1,9 @@
from .ledger import CardStatement, InstallmentPlan, Transaction, Transfer
from .provenance import AuditEvent, SourceRecord
from .reconciliation import ReconciliationCheckpoint
from .reference import Account, Category, Payee
__all__ = [
"Account", "AuditEvent", "CardStatement", "Category", "InstallmentPlan",
"Payee", "SourceRecord", "Transaction", "Transfer",
"Payee", "ReconciliationCheckpoint", "SourceRecord", "Transaction", "Transfer",
]

12
finance/models/reconciliation.py

@ -0,0 +1,12 @@
from django.db import models
from .base import TimestampedModel
from .reference import Account
class ReconciliationCheckpoint(TimestampedModel):
account = models.ForeignKey(Account, on_delete=models.PROTECT)
as_of_date = models.DateField()
institution_balance = models.DecimalField(max_digits=18, decimal_places=2)
ledger_balance_at_creation = models.DecimalField(max_digits=18, decimal_places=2)
difference_at_creation = models.DecimalField(max_digits=18, decimal_places=2)

96
finance/reconciliation.py

@ -0,0 +1,96 @@
"""Read models and shared calculations for reconciliation checkpoints."""
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from uuid import UUID
from django.db.models import DecimalField, F, Sum, Value, Window
from django.db.models.functions import Coalesce, RowNumber
from finance.models import ReconciliationCheckpoint, Transaction
_MONEY_FIELD = DecimalField(max_digits=24, decimal_places=2)
_ZERO = Decimal("0.00")
@dataclass(frozen=True)
class CheckpointStatus:
checkpoint_id: UUID
account_id: UUID
account_name: str
currency: str
as_of_date: date
created_at: datetime
institution_balance: Decimal
snapshot_ledger_balance: Decimal
snapshot_difference: Decimal
live_ledger_balance: Decimal
live_difference: Decimal
state: str
def ledger_balance_through(account_id: UUID, as_of_date: date) -> Decimal:
from finance.services.base import money
total = Transaction.objects.filter(
account_id=account_id,
posting_date__lte=as_of_date,
).aggregate(
total=Coalesce(
Sum("amount"),
Value(_ZERO),
output_field=_MONEY_FIELD,
)
)["total"]
return money(total, allow_zero=True)
def checkpoint_status(checkpoint: ReconciliationCheckpoint) -> CheckpointStatus:
account = checkpoint.account
live_ledger_balance = ledger_balance_through(checkpoint.account_id, checkpoint.as_of_date)
live_difference = checkpoint.institution_balance - live_ledger_balance
if live_difference == _ZERO:
state = "reconciled"
elif checkpoint.difference_at_creation == _ZERO:
state = "changed_since_reconciliation"
else:
state = "difference"
return CheckpointStatus(
checkpoint_id=checkpoint.pk,
account_id=account.pk,
account_name=account.name,
currency=account.currency,
as_of_date=checkpoint.as_of_date,
created_at=checkpoint.created_at,
institution_balance=checkpoint.institution_balance,
snapshot_ledger_balance=checkpoint.ledger_balance_at_creation,
snapshot_difference=checkpoint.difference_at_creation,
live_ledger_balance=live_ledger_balance,
live_difference=live_difference,
state=state,
)
def checkpoint_history(account_id: UUID) -> list[CheckpointStatus]:
checkpoints = ReconciliationCheckpoint.objects.filter(
account_id=account_id,
).select_related("account").order_by("-as_of_date", "-created_at", "-id")
return [checkpoint_status(checkpoint) for checkpoint in checkpoints]
def latest_checkpoint_statuses(currency: str) -> list[CheckpointStatus]:
checkpoints = ReconciliationCheckpoint.objects.filter(
account__archived=False,
account__currency=currency,
).annotate(
account_rank=Window(
expression=RowNumber(),
partition_by=[F("account_id")],
order_by=["-as_of_date", "-created_at", "-id"],
)
).filter(account_rank=1).select_related("account").order_by(
"-as_of_date", "-created_at", "-id"
)
return [checkpoint_status(checkpoint) for checkpoint in checkpoints]

3
finance/services/__init__.py

@ -2,6 +2,7 @@
from .base import Actor, DomainError
from .reference import create_account, create_payee, create_category, archive, create_statement, create_plan
from .posting import record_transaction, annotate_transaction, attach_source
from .reconciliation import create_reconciliation_checkpoint
from .transfers import create_transfer, match_transfer
__all__ = ["Actor", "DomainError", "create_account", "create_payee", "create_category", "archive", "create_statement", "create_plan", "record_transaction", "annotate_transaction", "attach_source", "create_transfer", "match_transfer"]
__all__ = ["Actor", "DomainError", "create_account", "create_payee", "create_category", "archive", "create_statement", "create_plan", "record_transaction", "annotate_transaction", "attach_source", "create_reconciliation_checkpoint", "create_transfer", "match_transfer"]

25
finance/services/reconciliation.py

@ -0,0 +1,25 @@
from finance.models import Account, ReconciliationCheckpoint
from finance.reconciliation import ledger_balance_through
from .base import lock, money, save, write
@write("reconciliation_checkpoint.created")
def create_reconciliation_checkpoint(*, account_id, as_of_date, institution_balance):
account = lock(Account, account_id, active=True)
institution_balance = money(institution_balance, allow_zero=True)
ledger_balance = ledger_balance_through(account.pk, as_of_date)
checkpoint = save(ReconciliationCheckpoint(
account=account,
as_of_date=as_of_date,
institution_balance=institution_balance,
ledger_balance_at_creation=ledger_balance,
difference_at_creation=money(institution_balance - ledger_balance, allow_zero=True),
))
return checkpoint, [
"account",
"as_of_date",
"institution_balance",
"ledger_balance_at_creation",
"difference_at_creation",
]

6
ingestion/parsers.py

@ -85,7 +85,9 @@ def _txt_items(text, profile, skipped):
partial = profile == 'OUROCARD_PARTIAL_V1'
date_size, body_size = (5, 43) if partial else (10, 37)
date_re = r'\d{2}/\d{2}' if partial else r'\d{2}\.\d{2}\.\d{4}'
regular = re.compile(rf'^({date_re})(.{{{body_size}}})([A-Z ]{{2}})\s+(\S+)\s+(\S+)\s*$')
regular = [re.compile(rf'^({date_re})(.{{{body_size}}})([A-Z ]{{2}})\s+(\S+)\s+(\S+)\s*$')]
if profile == 'OUROCARD_FULL_V1':
regular.append(re.compile(rf'^({date_re})(.{{33}})([A-Za-z ]{{7}})\s+(\S+)\s+(\S+)\s*$'))
payment = re.compile(rf'^({date_re})\s*(PGTO DEBITO CONTA.*?)200211\s+(\S+)\s+(\S+)\s*$')
section, section_count = '1', 0
for number, line in enumerate(text.splitlines(), 1):
@ -100,7 +102,7 @@ def _txt_items(text, profile, skipped):
skipped['balance'] += 1
elif match := payment.fullmatch(line):
yield ParsedItem(locator, section, match[1], match[2].strip(), match[3], match[4], True)
elif match := regular.fullmatch(line):
elif match := next((candidate for pattern in regular if (candidate := pattern.fullmatch(line))), None):
if _balance(match[2]):
skipped['balance'] += 1
else:

346
tests/finance/test_reconciliation.py

@ -0,0 +1,346 @@
"""Synthetic checks for immutable, audited reconciliation checkpoints."""
from datetime import date, datetime
from decimal import Decimal
from uuid import UUID
import pytest
from django.utils import timezone
from finance.models import Account, AuditEvent, ReconciliationCheckpoint
from finance.reconciliation import (
checkpoint_history,
latest_checkpoint_statuses,
ledger_balance_through,
)
from finance.services import (
Actor,
DomainError,
archive,
create_account,
create_reconciliation_checkpoint,
record_transaction,
)
pytestmark = pytest.mark.django_db(transaction=True)
ACTOR = Actor(kind="HUMAN", identifier="synthetic-local-owner")
AS_OF_DATE = date(2026, 9, 1)
def account(name="Synthetic reconciliation account"):
return create_account(name=name, kind="BANK", actor=ACTOR)
def post(account, amount, posting_date, kind="EXPENSE"):
return record_transaction(
account_id=account.pk,
amount=Decimal(amount),
kind=kind,
posting_date=posting_date,
actor=ACTOR,
)
def test_checkpoint_snapshots_inclusive_ledger_balance_and_audit_metadata():
bank = account()
post(bank, "-40.00", date(2026, 8, 31))
post(bank, "-2.50", AS_OF_DATE)
post(bank, "-7.50", date(2026, 9, 2))
checkpoint = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
assert checkpoint.institution_balance == Decimal("0.00")
assert checkpoint.ledger_balance_at_creation == Decimal("-42.50")
assert checkpoint.difference_at_creation == Decimal("42.50")
event = AuditEvent.objects.get(
operation="reconciliation_checkpoint.created",
target_id=checkpoint.pk,
)
assert event.outcome == "SUCCESS"
assert event.actor_kind == "HUMAN"
assert event.actor_identifier == "synthetic-local-owner"
assert event.changed_fields == [
"account",
"as_of_date",
"institution_balance",
"ledger_balance_at_creation",
"difference_at_creation",
]
def test_ledger_balance_through_returns_zero_and_includes_opening_entries():
bank = account()
assert ledger_balance_through(bank.pk, AS_OF_DATE) == Decimal("0.00")
post(bank, "100.00", date(2026, 8, 31), kind="OPENING")
post(bank, "-40.00", AS_OF_DATE)
post(bank, "-7.50", date(2026, 9, 2))
assert ledger_balance_through(bank.pk, AS_OF_DATE) == Decimal("60.00")
def test_checkpoint_correction_retains_same_date_history():
bank = account()
post(bank, "-40.00", AS_OF_DATE)
first = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-39.00"),
actor=ACTOR,
)
correction = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-40.00"),
actor=ACTOR,
)
assert first.pk != correction.pk
assert ReconciliationCheckpoint.objects.filter(
account=bank,
as_of_date=AS_OF_DATE,
).count() == 2
assert correction.ledger_balance_at_creation == Decimal("-40.00")
assert correction.difference_at_creation == Decimal("0.00")
def test_archived_account_rejects_checkpoint_without_creating_one():
bank = account()
archive(bank, actor=ACTOR)
checkpoint_count = ReconciliationCheckpoint.objects.count()
audit_count = AuditEvent.objects.count()
with pytest.raises(DomainError, match="archived_reference"):
create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
assert ReconciliationCheckpoint.objects.count() == checkpoint_count
failed = AuditEvent.objects.exclude(operation="archive").latest("created_at")
assert AuditEvent.objects.count() == audit_count + 1
assert failed.operation == "reconciliation_checkpoint.created"
assert failed.outcome == "FAILURE"
assert failed.error_code == "archived_reference"
@pytest.mark.parametrize(
(
"institution_balance",
"later_amount",
"expected_live_ledger_balance",
"expected_state",
"expected_live_difference",
),
[
("-40.00", None, Decimal("-40.00"), "reconciled", Decimal("0.00")),
("-39.00", None, Decimal("-40.00"), "difference", Decimal("1.00")),
("-40.00", "-2.00", Decimal("-42.00"), "changed_since_reconciliation", Decimal("2.00")),
],
)
def test_checkpoint_history_reports_live_reconciliation_state(
institution_balance,
later_amount,
expected_live_ledger_balance,
expected_state,
expected_live_difference,
):
bank = account(name=f"Synthetic status {expected_state}")
post(bank, "-40.00", AS_OF_DATE)
checkpoint = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal(institution_balance),
actor=ACTOR,
)
if later_amount is not None:
post(bank, later_amount, AS_OF_DATE)
[status] = checkpoint_history(bank.pk)
assert status.checkpoint_id == checkpoint.pk
assert status.snapshot_ledger_balance == Decimal("-40.00")
assert status.live_ledger_balance == expected_live_ledger_balance
assert status.live_difference == expected_live_difference
assert status.state == expected_state
def test_checkpoint_history_retains_same_date_corrections_newest_first():
bank = account()
post(bank, "-40.00", AS_OF_DATE)
first = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-39.00"),
actor=ACTOR,
)
correction = create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-40.00"),
actor=ACTOR,
)
history = checkpoint_history(bank.pk)
assert [status.checkpoint_id for status in history] == [correction.pk, first.pk]
assert history[0].snapshot_difference == Decimal("0.00")
assert history[1].snapshot_difference == Decimal("1.00")
def test_latest_checkpoint_statuses_select_by_as_of_date_then_creation_for_active_currency():
by_date = account(name="Synthetic latest by date")
post(by_date, "-40.00", AS_OF_DATE)
newer_as_of_date = create_reconciliation_checkpoint(
account_id=by_date.pk,
as_of_date=date(2026, 9, 2),
institution_balance=Decimal("-40.00"),
actor=ACTOR,
)
create_reconciliation_checkpoint(
account_id=by_date.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-39.00"),
actor=ACTOR,
)
by_creation = account(name="Synthetic latest by creation")
post(by_creation, "-20.00", AS_OF_DATE)
create_reconciliation_checkpoint(
account_id=by_creation.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-19.00"),
actor=ACTOR,
)
correction = create_reconciliation_checkpoint(
account_id=by_creation.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-20.00"),
actor=ACTOR,
)
archived = account(name="Synthetic archived status")
post(archived, "-5.00", AS_OF_DATE)
create_reconciliation_checkpoint(
account_id=archived.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-5.00"),
actor=ACTOR,
)
archive(archived, actor=ACTOR)
usd = create_account(
name="Synthetic USD status", kind="BANK", currency="USD", actor=ACTOR
)
post(usd, "-7.00", AS_OF_DATE)
create_reconciliation_checkpoint(
account_id=usd.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("-7.00"),
actor=ACTOR,
)
statuses = latest_checkpoint_statuses("BRL")
assert {status.account_id: status.checkpoint_id for status in statuses} == {
by_date.pk: newer_as_of_date.pk,
by_creation.pk: correction.pk,
}
def test_latest_checkpoint_statuses_order_selected_accounts_by_checkpoint_recency():
oldest_account = Account.objects.create(
id=UUID("00000000-0000-0000-0000-000000000001"),
name="Synthetic oldest account",
kind="BANK",
)
older_same_date_account = Account.objects.create(
id=UUID("00000000-0000-0000-0000-000000000002"),
name="Synthetic older same-date account",
kind="BANK",
)
newest_account = Account.objects.create(
id=UUID("00000000-0000-0000-0000-000000000003"),
name="Synthetic newest account",
kind="BANK",
)
oldest = create_reconciliation_checkpoint(
account_id=oldest_account.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
older_same_date = create_reconciliation_checkpoint(
account_id=older_same_date_account.pk,
as_of_date=date(2026, 9, 2),
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
newest = create_reconciliation_checkpoint(
account_id=newest_account.pk,
as_of_date=date(2026, 9, 2),
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
ReconciliationCheckpoint.objects.filter(pk=oldest.pk).update(
created_at=timezone.make_aware(datetime(2026, 9, 1, 9, 0))
)
ReconciliationCheckpoint.objects.filter(pk=older_same_date.pk).update(
created_at=timezone.make_aware(datetime(2026, 9, 1, 10, 0))
)
ReconciliationCheckpoint.objects.filter(pk=newest.pk).update(
created_at=timezone.make_aware(datetime(2026, 9, 1, 11, 0))
)
statuses = latest_checkpoint_statuses("BRL")
assert [status.checkpoint_id for status in statuses] == [
newest.pk,
older_same_date.pk,
oldest.pk,
]
def test_latest_checkpoint_statuses_order_equal_recency_by_checkpoint_id_descending():
first_account = Account.objects.create(
id=UUID("00000000-0000-0000-0000-000000000004"),
name="Synthetic first equal-recency account",
kind="BANK",
)
second_account = Account.objects.create(
id=UUID("00000000-0000-0000-0000-000000000005"),
name="Synthetic second equal-recency account",
kind="BANK",
)
first = create_reconciliation_checkpoint(
account_id=first_account.pk,
as_of_date=date(2026, 9, 2),
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
second = create_reconciliation_checkpoint(
account_id=second_account.pk,
as_of_date=date(2026, 9, 2),
institution_balance=Decimal("0.00"),
actor=ACTOR,
)
same_creation_time = timezone.make_aware(datetime(2026, 9, 1, 12, 0))
ReconciliationCheckpoint.objects.filter(pk__in=[first.pk, second.pk]).update(
created_at=same_creation_time
)
statuses = latest_checkpoint_statuses("BRL")
assert [status.checkpoint_id for status in statuses] == sorted(
[first.pk, second.pk], reverse=True
)

12
tests/test_import_parsers.py

@ -20,6 +20,10 @@ def card_line(day='15.02.2026', body='LOJA INVENTADA', debit='12,34', credit='0,
return day + body.ljust(43 if partial else 37) + 'BR' + f' {debit} {credit}'
def alternate_full_card_line(day='15.02.2026', body='LOJA EXTERIOR', country='ESTADOS', debit='12,34', credit='0,00'):
return day + body.ljust(33) + country.ljust(7) + f' {debit} {credit}'
def card(text, partial=False, context=None):
return normalize(parse_document(text.encode(), 'OUROCARD_PARTIAL_V1' if partial else 'OUROCARD_FULL_V1', context))
@ -60,6 +64,14 @@ def test_full_card_sections_installments_credit_and_payment():
assert rows[3].description == 'unknown text'
@pytest.mark.parametrize('country', ['ESTADOS', 'Brasil'])
def test_full_card_accepts_extended_country_layout(country):
row = card(alternate_full_card_line(country=country))[0]
assert row.errors == ()
assert row.description == 'LOJA EXTERIOR'
assert row.amount == '-12.34'
def test_card_ambiguity_bad_shapes_and_installments_block():
assert 'AMBIGUOUS_AMOUNT' in card(card_line(debit='2,00', credit='1,00'))[0].errors
assert 'INVALID_ROW' in card('15.02.2026 MALFORMED')[0].errors

60
tests/test_mvp_acceptance.py

@ -2,6 +2,7 @@
import json
from contextlib import contextmanager
from datetime import date
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace
@ -18,8 +19,8 @@ from core.backup import (
verify_backup,
)
from finance.exports import build_export
from finance.models import SourceRecord, Transaction
from finance.services import Actor, create_account, create_category
from finance.models import ReconciliationCheckpoint, SourceRecord, Transaction
from finance.services import Actor, create_account, create_category, record_transaction
from ingestion import services as import_services
@ -28,6 +29,61 @@ ACTOR = Actor(kind="AGENT", identifier="synthetic-mvp-acceptance")
FIXTURE = Path("tests/fixtures/synthetic/ingestion/bank.csv")
def test_owner_reconciliation_workflow_reports_backdated_activity():
account = create_account(
name="Synthetic acceptance reconciliation account",
kind="BANK",
currency="BRL",
actor=ACTOR,
)
record_transaction(
account_id=account.pk,
amount=Decimal("-50.00"),
posting_date=date(2026, 4, 10),
kind="EXPENSE",
actor=ACTOR,
)
browser = Client()
checkpoint = {
"account": account.pk,
"as_of_date": "2026-04-30",
"institution_balance": "-50.00",
}
preview = browser.post(
reverse("reconciliation-index"),
{"action": "preview", **checkpoint},
)
assert preview.status_code == 200
assert "Preview" in preview.content.decode()
assert "0.00 BRL" in preview.content.decode()
assert ReconciliationCheckpoint.objects.count() == 0
confirmed = browser.post(
reverse("reconciliation-index"),
{"action": "confirm", "preview_token": preview.context["preview_token"], **checkpoint},
)
assert confirmed.status_code == 302
assert confirmed.url == reverse("reconciliation-history", args=[account.pk])
assert ReconciliationCheckpoint.objects.count() == 1
record_transaction(
account_id=account.pk,
amount=Decimal("-5.00"),
posting_date=date(2026, 4, 15),
kind="EXPENSE",
actor=ACTOR,
)
dashboard = browser.get(
reverse("home"),
{"month": "2026-04", "currency": "BRL"},
)
assert dashboard.status_code == 200
assert "Synthetic acceptance reconciliation account" in dashboard.content.decode()
assert "Changed since reconciliation" in dashboard.content.decode()
def test_synthetic_import_to_report_export_and_backup_workflow(
tmp_path, settings, monkeypatch
):

320
tests/test_reconciliation_views.py

@ -0,0 +1,320 @@
"""Synthetic end-to-end checks for the owner reconciliation pages."""
from datetime import date, datetime
from decimal import Decimal
import re
import pytest
from django.test import Client
from django.urls import reverse
from django.utils import timezone
from finance.models import AuditEvent, ReconciliationCheckpoint
from finance.services import (
Actor,
archive,
create_account,
create_reconciliation_checkpoint,
record_transaction,
)
pytestmark = pytest.mark.django_db(transaction=True)
SETUP_ACTOR = Actor(kind="AGENT", identifier="synthetic-reconciliation-view-test")
AS_OF_DATE = date(2026, 9, 1)
def account(name="Synthetic reconciliation bank"):
return create_account(
name=name,
kind="BANK",
currency="BRL",
actor=SETUP_ACTOR,
)
def post(target, amount, *, posting_date=AS_OF_DATE, kind="EXPENSE"):
return record_transaction(
account_id=target.pk,
amount=Decimal(amount),
posting_date=posting_date,
kind=kind,
actor=SETUP_ACTOR,
)
def checkpoint(target, institution_balance):
return create_reconciliation_checkpoint(
account_id=target.pk,
as_of_date=AS_OF_DATE,
institution_balance=Decimal(institution_balance),
actor=SETUP_ACTOR,
)
def preview_token(response):
match = re.search(r'name="preview_token" value="([^"]+)"', response.content.decode())
return match.group(1) if match else ""
@pytest.mark.parametrize("field", ["account", "as_of_date", "institution_balance"])
def test_confirm_requires_new_preview_after_input_changes(client, field):
bank = account()
other = account("Synthetic other reconciliation bank")
values = {
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
url = reverse("reconciliation-index")
preview = client.post(url, {"action": "preview", **values})
changed = {**values, field: {
"account": other.pk,
"as_of_date": "2026-09-02",
"institution_balance": "12.50",
}[field]}
response = client.post(url, {
"action": "confirm", "preview_token": preview_token(preview), **changed,
})
assert ReconciliationCheckpoint.objects.count() == 0
assert response.status_code == 200
assert response.context["form"].non_field_errors()
assert 'name="action" value="confirm"' not in response.content.decode()
refreshed = client.post(url, {"action": "preview", **changed})
confirmed = client.post(url, {
"action": "confirm", "preview_token": preview_token(refreshed), **changed,
})
assert confirmed.status_code == 302
saved = ReconciliationCheckpoint.objects.get()
assert str(saved.account_id) == str(changed["account"])
assert saved.as_of_date.isoformat() == changed["as_of_date"]
assert saved.institution_balance == Decimal(changed["institution_balance"])
@pytest.mark.parametrize("token_kind", ["missing", "tampered"])
def test_confirm_rejects_missing_or_tampered_preview_token(client, token_kind):
bank = account()
values = {
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
url = reverse("reconciliation-index")
preview = client.post(url, {"action": "preview", **values})
token = "" if token_kind == "missing" else preview_token(preview) + "tampered"
response = client.post(url, {
"action": "confirm", "preview_token": token, **values,
})
assert ReconciliationCheckpoint.objects.count() == 0
assert response.status_code == 200
assert response.context["form"].non_field_errors()
def test_history_exposes_creation_timestamps_for_same_date_corrections(client):
bank = account()
first = checkpoint(bank, "0.00")
correction = checkpoint(bank, "1.00")
first_time = timezone.make_aware(datetime(2026, 9, 2, 10, 11, 12))
correction_time = timezone.make_aware(datetime(2026, 9, 2, 10, 11, 13))
ReconciliationCheckpoint.objects.filter(pk=first.pk).update(created_at=first_time)
ReconciliationCheckpoint.objects.filter(pk=correction.pk).update(created_at=correction_time)
response = client.get(reverse("reconciliation-history", args=[bank.pk]))
body = response.content.decode()
assert "Created at" in body
assert "2026-09-02 10:11:12 UTC" in body
assert "2026-09-02 10:11:13 UTC" in body
assert body.index("10:11:13 UTC") < body.index("10:11:12 UTC")
assert [row.created_at for row in response.context["history"]] == [
correction_time, first_time,
]
def test_reconciliation_is_linked_and_landing_form_lists_only_active_accounts(client):
active = account("Synthetic active bank")
archived = account("Synthetic archived bank")
checkpoint(active, "0.00")
archive(archived, actor=SETUP_ACTOR)
navigation_page = client.get(reverse("transaction-list"))
response = client.get(reverse("reconciliation-index"))
assert navigation_page.status_code == 200
assert f'href="{reverse("reconciliation-index")}">Reconciliation</a>' in navigation_page.content.decode()
assert response.status_code == 200
body = response.content.decode()
assert "<h1>Reconciliation</h1>" in body
assert f'<option value="{active.pk}">Synthetic active bank</option>' in body
assert str(archived.pk) not in body
assert "Latest reconciliation status" in body
assert "Reconciled" in body
assert reverse("reconciliation-history", args=[active.pk]) in body
def test_preview_renders_ledger_balance_and_difference_without_writing(client):
bank = account()
post(bank, "100.00", kind="OPENING", posting_date=date(2026, 8, 31))
post(bank, "-40.00")
response = client.post(
reverse("reconciliation-index"),
{
"action": "preview",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 200
body = response.content.decode()
assert "Preview" in body
assert "Ledger balance" in body and "60.00 BRL" in body
assert "Difference" in body and "-60.00 BRL" in body
assert 'name="action" value="confirm"' in body
assert ReconciliationCheckpoint.objects.count() == 0
def test_confirm_revalidates_and_service_snapshots_fresh_ledger_then_redirects(client):
bank = account()
post(bank, "-40.00")
preview = client.post(
reverse("reconciliation-index"),
{
"action": "preview",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "-40.00",
},
)
assert preview.status_code == 200
assert "0.00 BRL" in preview.content.decode()
post(bank, "-2.00")
response = client.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "-40.00",
"ledger_balance": "999999.00",
"difference": "999999.00",
"preview_token": preview_token(preview),
},
)
saved = ReconciliationCheckpoint.objects.get()
assert response.status_code == 302
assert response.url == reverse("reconciliation-history", args=[bank.pk])
assert saved.ledger_balance_at_creation == Decimal("-42.00")
assert saved.difference_at_creation == Decimal("2.00")
event = AuditEvent.objects.get(
operation="reconciliation_checkpoint.created",
target_id=saved.pk,
)
assert (event.actor_kind, event.actor_identifier) == ("HUMAN", "local-owner")
@pytest.mark.parametrize(
("field", "value", "expected_error"),
[
("as_of_date", "not-a-date", "Enter a valid date."),
("institution_balance", "not-money", "Enter a number."),
("institution_balance", "1.001", "Ensure that there are no more than 2 decimal places."),
],
)
def test_malformed_values_show_field_errors_without_writing(
client, field, value, expected_error
):
bank = account()
payload = {
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
}
payload[field] = value
response = client.post(reverse("reconciliation-index"), payload)
assert response.status_code == 200
assert expected_error in response.content.decode()
assert field in response.context["form"].errors
assert ReconciliationCheckpoint.objects.count() == 0
def test_reconciliation_confirmation_requires_csrf():
bank = account()
guarded = Client(enforce_csrf_checks=True)
response = guarded.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": bank.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 403
assert ReconciliationCheckpoint.objects.count() == 0
def test_archived_account_confirmation_fails_safely_without_writing(client):
archived = account()
archive(archived, actor=SETUP_ACTOR)
response = client.post(
reverse("reconciliation-index"),
{
"action": "confirm",
"account": archived.pk,
"as_of_date": "2026-09-01",
"institution_balance": "0.00",
},
)
assert response.status_code == 200
body = response.content.decode()
assert "Select a valid choice." in body
assert "synthetic_private" not in body
assert ReconciliationCheckpoint.objects.count() == 0
def test_history_keeps_corrected_same_date_snapshots_and_shows_live_drift(client):
bank = account()
post(bank, "-40.00")
first = checkpoint(bank, "-39.25")
correction = checkpoint(bank, "-40.00")
post(bank, "-2.00")
archive(bank, actor=SETUP_ACTOR)
response = client.get(reverse("reconciliation-history", args=[bank.pk]))
assert response.status_code == 200
body = response.content.decode()
assert "Synthetic reconciliation bank" in body
assert "Archived" in body
assert "Institution balance" in body
assert "Snapshot ledger balance" in body
assert "Snapshot difference" in body
assert "Live ledger balance" in body
assert "Live difference" in body
assert "Changed since reconciliation" in body
assert "Difference" in body
assert "-42.00 BRL" in body
assert "2.00 BRL" in body
assert "2.75 BRL" in body
assert [row.checkpoint_id for row in response.context["history"]] == [
correction.pk,
first.pk,
]

37
tests/test_report_views.py

@ -6,9 +6,11 @@ from django.urls import reverse
from django.utils import timezone
from finance.models import Account, Category, Transaction
from finance.services import Actor, create_reconciliation_checkpoint, record_transaction
pytestmark = pytest.mark.django_db
SETUP_ACTOR = Actor(kind="AGENT", identifier="synthetic-report-view-test")
def make_account(name="Synthetic Bank", *, currency="BRL"):
@ -121,3 +123,38 @@ def test_negative_spending_and_balance_keep_their_sign_and_are_labeled(client):
assert "-25.00" in body
assert "Balance (negative)" in body
assert "-75.00" in body
@pytest.mark.django_db(transaction=True)
def test_dashboard_displays_current_reconciliation_status_after_backdated_activity(client):
bank = make_account("Synthetic reconciled bank")
record_transaction(
account_id=bank.pk,
amount=Decimal("-40.00"),
posting_date=date(2026, 1, 15),
kind="EXPENSE",
actor=SETUP_ACTOR,
)
create_reconciliation_checkpoint(
account_id=bank.pk,
as_of_date=date(2026, 1, 31),
institution_balance=Decimal("-40.00"),
actor=SETUP_ACTOR,
)
record_transaction(
account_id=bank.pk,
amount=Decimal("-2.50"),
posting_date=date(2026, 1, 20),
kind="EXPENSE",
actor=SETUP_ACTOR,
)
response = client.get(reverse("home"), {"month": "2026-01", "currency": "BRL"})
assert response.status_code == 200
body = response.content.decode()
assert "Reconciliation status" in body
assert "Synthetic reconciled bank" in body
assert "Changed since reconciliation" in body
assert "2.50 BRL" in body
assert reverse("reconciliation-history", args=[bank.pk]) in body

Loading…
Cancel
Save