Design a Tunable Recurring-Transaction Detector

Problem Statement

Corporate cardholders frequently incur recurring SaaS and vendor subscriptions. However, real-world transaction logs are messy: descriptions vary (e.g., GITHUB* 123 vs GITHUB INC), billing intervals drift due to weekends and leap days (e.g., 28 to 33 days for monthly), and prices fluctuate due to seat changes, tier upgrades, or tax adjustments.

Design a class RecurringDetector that ingests a user's transaction history and outputs detected recurring subscription streams with explainable metadata.

Requirements

1. Merchant Normalization: Normalize raw transaction descriptions by stripping non-alphanumeric trailing tokens (such as IDs, reference codes, state codes, and asterisks) to group related transactions.
2. Cadence Classification: For a candidate group of transactions from the same merchant (minimum 3 transactions), determine if they follow a regular cadence:
   • WEEKLY: Average interval between consecutive occurrences is approximately 7 days (within interval tolerance t_days).
   • MONTHLY: Average interval is approximately 30 days (28–33 days acceptable within tolerance).
   • ANNUAL: Average interval is approximately 365 days (360–370 days acceptable within tolerance).
3. Amount Variance: Transactions within a detected stream should be within an amount tolerance t_amount_pct relative to the stream's median amount, but should tolerate at most one legitimate step-change (e.g., a plan upgrade or price increase).
4. Duplicate Deduplication: Ignore duplicate transactions from the same merchant with identical amounts occurring within 24 hours of each other (keep only the first).
5. Explainability: Each detected stream must return the merchant key, detected cadence, average amount, estimated next charge date, and a confidence score between 0.0 and 1.0 based on regularity.

Interface

from dataclasses import dataclass
from typing import List, Optional
from datetime import date

@dataclass
class Transaction:
    id: str
    description: str
    amount_cents: int
    date: str  # "YYYY-MM-DD"

@dataclass
class RecurringStream:
    merchant: str
    cadence: str           # "WEEKLY", "MONTHLY", "ANNUAL"
    typical_amount_cents: int
    next_expected_date: str
    confidence: float      # 0.0 to 1.0
    transaction_ids: List[str]

class RecurringDetector:
    def __init__(self, t_days: int = 3, t_amount_pct: float = 0.15):
        """
        t_days: allowed drift in days from expected interval.
        t_amount_pct: allowed fractional deviation from median (e.g., 0.15 for 15%).
        """
        pass

    def detect(self, transactions: List[Transaction]) -> List[RecurringStream]:
        pass

Constraints

• Number of transactions N <= 100,000
• Transactions can be provided in arbitrary order.
• Amounts are positive integers representing cents (1 <= amount_cents <= 100,000,000).
• Time complexity should be O(N log N) or better.

Example

txs = [
    Transaction("1", "SLACK TECHNOLOGIES *441", 1200, "2023-01-15"),
    Transaction("2", "SLACK TECHNOLOGIES *998", 1200, "2023-02-15"),
    Transaction("3", "SLACK INC", 1200, "2023-03-16"),
    Transaction("4", "SLACK INC", 1200, "2023-04-15"),
]
detector = RecurringDetector(t_days=3, t_amount_pct=0.10)
streams = detector.detect(txs)
# Expected:
# [RecurringStream(merchant="SLACK", cadence="MONTHLY", typical_amount_cents=1200,
#                  next_expected_date="2023-05-15", confidence=0.95, transaction_ids=["1","2","3","4"])]