Sign inorSign up
55 lines
jobs/transaction_summary.py
Filters and aggregates completed transactions for a single date.
# Daily transaction summary job for the finance reporting pipeline.
import logging
from decimal import Decimal
from typing import List
 
logger = logging.getLogger(__name__)
 
# Type alias: transaction dict with tx_id (str), date (str YYYY-MM-DD),
# amount (str decimal), status ('completed' | 'failed' | 'pending').
Transaction = dict
 
 
def build_daily_summary(transactions: List[Transaction], date: str) -> dict:
    """Compute a summary of completed transactions for the given date.
 
    Filters to transactions where date matches and status is 'completed'.
    Returns a dict with count, total_amount, and average_amount.
    When count is zero, average_amount is Decimal('0.00').
 
    Parameters
    ----------
    transactions : list of Transaction
        Full transaction list for the period; non-matching rows are ignored.
    date : str
        Target date in YYYY-MM-DD format.
 
    Returns
    -------
    dict
        Summary with keys: date (str), count (int),
        total_amount (Decimal), average_amount (Decimal).
    """
    daily = [
        t for t in transactions
        if t["date"] == date and t["status"] == "completed"
    ]
    count = len(daily)
    total_amount = sum(
        (Decimal(t["amount"]) for t in daily), Decimal("0.00")
    )
    average_amount = total_amount / count
 
    logger.info(
        "transaction_summary: date=%s count=%d total=%s average=%s",
        date,
        count,
        total_amount,
        average_amount,
    )
    return {
        "date": date,
        "count": count,
        "total_amount": total_amount,
        "average_amount": average_amount,
    }