Detecting cost transfer anomalies in grant ledgers

Problem statement #

You need a deterministic Python rule engine that scans a stream of ledger journal entries, identifies cost transfers that carry audit risk — those posted more than 90 days after the original charge, those landing on an award near or past its end date, frequent or round-dollar shuffles between awards, and entries hitting unallowable object codes — and routes every suspect entry to a compliance quarantine with a specific reason code, never auto-approving, so that re-running the scan over the same ledger produces the same flags every time.

This task sits under Indirect Cost Recovery Reconciliation, part of the broader Compliance Reporting & Budget Reconciliation practice. Cost transfers are the mechanism by which the modified total direct cost base quietly shifts after the fact, so screening them protects the same base that the companion F&A recovery reconciliation depends on. The engine detects and routes; it never decides that a transfer is acceptable — that judgment belongs to a compliance officer.

Prerequisites #

• Python 3.10+ for modern type hints and structural pattern matching.
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] for the idempotent quarantine write. Date handling uses only the standard-library datetime module with explicit UTC. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • LEDGER_DB_URL — the SQLAlchemy connection string for the ledger and quarantine store.
• Policy config: a version-controlled list of unallowable object codes and the institutional cost-transfer policy (the 90-day window and the required-justification rule), maintained alongside your University Policy Mapping Frameworks. Journal entries are assumed already mapped and validated upstream by the Grant Lifecycle Architecture Design layer.

Step-by-step implementation #

The engine enforces the flow below: each journal entry is normalized to UTC dates, its age is computed as days_late, a fixed sequence of rules is evaluated, the first matching rule assigns a reason code, and any flagged entry is upserted into the quarantine on the je_id key. No entry is ever approved by the engine — an entry that trips no rule is simply left in place for normal posting.

Step 1 — Load and normalize journal entries #

Read each journal entry and normalize both the original charge date and the posting date to UTC immediately. A cost transfer’s age drives the primary rule, so a naïve local-vs-UTC mismatch on the two dates would corrupt days_late before any rule runs.

from datetime import datetime, timezone, date
from dataclasses import dataclass
from decimal import Decimal


@dataclass(frozen=True)
class JournalEntry:
    je_id: str
    orig_date: datetime      # when the original charge posted
    post_date: datetime      # when this transfer posts
    amount: Decimal
    from_award: str
    to_award: str
    object_code: str
    to_award_end: date       # end date of the receiving award


def to_utc(value: datetime) -> datetime:
    """Force any datetime to UTC so date arithmetic is comparable."""
    if value.tzinfo is None:
        # A naive timestamp is assumed UTC; never subtract a naive from an aware one.
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def days_late(entry: JournalEntry) -> int:
    """Whole days between the original charge and this transfer, UTC-normalized."""
    delta = to_utc(entry.post_date).date() - to_utc(entry.orig_date).date()
    return delta.days

Step 2 — Declare the rules as data #

Each rule is a named predicate with a reason code. Declaring them in an ordered list — rather than a nest of if branches — keeps the engine deterministic and makes the policy auditable: a compliance officer can read the rule set without reading control flow.

LATE_THRESHOLD_DAYS = 90        # institutional 90-day cost-transfer policy
END_WINDOW_DAYS = 30            # "near end of award" window
ROUND_DOLLAR = Decimal("1000")  # transfers that are exact multiples look suspect

UNALLOWABLE_CODES = {
    "5800",   # alcohol
    "5810",   # entertainment
    "5820",   # fines and penalties
    "5830",   # lobbying
}


def rule_over_90_days(e: JournalEntry) -> bool:
    # 2 CFR 200.405 allocability: a transfer long after the charge is hard to justify.
    return days_late(e) > LATE_THRESHOLD_DAYS


def rule_near_award_end(e: JournalEntry) -> bool:
    # Transfers ONTO an award as it closes are a classic overspend-dumping signal.
    # A negative "remaining" means the award has already expired — also a flag.
    remaining = (e.to_award_end - to_utc(e.post_date).date()).days
    return remaining <= END_WINDOW_DAYS


def rule_round_dollar(e: JournalEntry) -> bool:
    # Round-dollar transfers rarely reflect an actual mischarge being corrected.
    return e.amount >= ROUND_DOLLAR and e.amount % ROUND_DOLLAR == 0


def rule_unallowable_code(e: JournalEntry) -> bool:
    # 2 CFR 200.403 allowability: some object codes are never chargeable to awards.
    return e.object_code in UNALLOWABLE_CODES


# Order matters: the first match wins, so the most serious rules come first.
RULES: list[tuple[str, callable]] = [
    ("UNALLOWABLE_OBJECT_CODE", rule_unallowable_code),
    ("TRANSFER_ONTO_EXPIRING_AWARD", rule_near_award_end),
    ("OVER_90_DAYS_LATE", rule_over_90_days),
    ("ROUND_DOLLAR_TRANSFER", rule_round_dollar),
]

Step 3 — Evaluate and assign a reason code #

The evaluator returns the first tripped rule’s reason code, or None for a clean entry. Because the rule list is fixed and ordered, the same entry always yields the same reason code — the engine is deterministic by construction.

def evaluate(entry: JournalEntry) -> str | None:
    """Return the reason code of the first rule the entry trips, or None if clean."""
    for reason_code, predicate in RULES:
        if predicate(entry):
            return reason_code
    return None

Step 4 — Route suspects to quarantine, never auto-approve #

A flagged entry is upserted into the quarantine on its je_id, so re-scanning the same ledger updates the existing quarantine row rather than inserting a duplicate. A clean entry is explicitly not written anywhere by this engine — the absence of a quarantine row is what “allowed to post” means, and the engine never stamps an approval.

import logging
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

logger = logging.getLogger("cost_transfer_scan")


def scan(entries: list[JournalEntry], session: Session) -> dict[str, int]:
    """Flag suspect cost transfers to quarantine. Idempotent on je_id."""
    stats = {"scanned": 0, "flagged": 0}
    for entry in entries:
        stats["scanned"] += 1
        reason = evaluate(entry)
        if reason is None:
            continue                       # clean: left in place, never approved
        stats["flagged"] += 1
        logger.warning("cost-transfer flagged je=%s reason=%s days_late=%s",
                       entry.je_id, reason, days_late(entry))
        row = {
            "je_id": entry.je_id,
            "orig_date": to_utc(entry.orig_date),
            "post_date": to_utc(entry.post_date),
            "days_late": days_late(entry),
            "amount": entry.amount,
            "from_award": entry.from_award,
            "to_award": entry.to_award,
            "object_code": entry.object_code,
            "reason_code": reason,
            "flag": "QUARANTINED",
        }
        stmt = pg_insert(CostTransferQuarantine).values(**row)
        stmt = stmt.on_conflict_do_update(
            index_elements=["je_id"],
            set_={k: stmt.excluded[k] for k in (
                "days_late", "amount", "reason_code", "flag")},
        )
        session.execute(stmt)
    session.commit()
    return stats

The on_conflict_do_update on je_id is the idempotency guarantee: re-scanning a ledger that has not changed produces the same quarantine rows with the same reason codes, and a corrected entry re-evaluates in place. Resolved quarantine items feed the audit record built by Immutable Audit Report Generation, and the salary transfers among them intersect with Effort Reporting & Certification.

Schema and field reference #

Verification #

1. Reproduce the flags. Run scan twice over the same entries; the quarantine row count must not change and every reason_code must be identical. A changed reason code on a second pass means a rule reads mutable or non-UTC state.
2. Recompute days_late. For a quarantined row, recompute to_utc(post_date).date() − to_utc(orig_date).date() and confirm it matches the stored days_late. A mismatch points at a timezone bug in ingestion.
3. Rule-boundary check. Feed an entry at exactly 90 days and confirm it is not flagged for lateness (the rule is strictly greater than 90), and one at 91 days and confirm it is. Repeat for the award-end window.
4. Clean-entry audit. Confirm that an entry tripping no rule leaves no quarantine row and receives no approval stamp — “allowed to post” is the absence of a flag, not a positive approval.

Troubleshooting #

• Timezone mismatch on orig vs post date inflating days_late. If the original charge date is stored in local time and the posting date in UTC (or vice versa), subtracting them can add or drop a day and push a compliant 89-day transfer over the 90-day line, or hide a genuinely late one. Normalize both dates with to_utc before any subtraction, and never subtract a naive datetime from an aware one — Python raises, and silencing that error by stripping tzinfo reintroduces the bug.
• Missing written justification. The 90-day rule catches lateness, but a transfer inside the window still requires a written, PI-approved justification under institutional policy. The engine cannot read a justification memo, so treat a flagged-and-cleared transfer as unresolved until the justification is attached in the quarantine record; never let the mere passage of review time clear a flag.
• Transfer to an already-expired award. A charge moved onto an award whose end date has passed is unallowable regardless of days_late. The rule_near_award_end predicate treats a negative “days remaining” as a flag precisely for this case, so confirm the receiving award’s to_award_end is populated — a null end date silently disables the check and lets post-expiration dumping through.

Frequently asked questions #

Related #

• Up to the parent topic: Indirect Cost Recovery Reconciliation
• Reconciling F&A Indirect-Cost Recovery Rates in Python — the base these transfers distort
• Compliance Reporting & Budget Reconciliation — the parent practice this screen feeds
• University Policy Mapping Frameworks — where the unallowable-code list and cost-transfer policy live