Improve detector schema + add API tester

README.md

@@ -33,10 +33,10 @@ Quick Start
 
 2. Copy `.env.example` to `.env` and set the Mongo connection string if you do not want to rely on the baked-in default.
 
-3. Run the detector (the default config scans the last 48 h of data and writes suggestions only):
+3. Run the detector (the default config scans the last 48 h of data and writes suggestions only). For the historical `transactions` collection you may want to bump the lookback window:
 
    ```
-   python3 -m src.main --minutes 12 --lookback-hours 72
+   python3 -m src.main --minutes 30 --lookback-hours 720
    ```
 
    You will see log lines such as:
@@ -63,7 +63,7 @@ Endpoints:
 Collections
 -----------
 
-* `expenses`: source data. The detector expects fields such as `_id`, `amount`, `currency`, `merchant`, `expense_time`.
+* `transactions` (default): source data. The detector automatically maps the `date`/`createdAt` timestamp and `note`/`paymentType` merchant fields so you still get near-duplicate detection without reshaping your documents. Entries are only compared if they belong to the same `user`.
 * `merchant_aliases`: optional alias definitions (`name`, `aliases`).
 * `merge_suggestions`: the service writes documents shaped as:
 
@@ -94,12 +94,26 @@ All tunables live in `src/config.py`. Environment variables take precedence, so
 | --- | --- | --- |
 | `MONGO_URI` | Mongo connection string | Provided URI |
 | `MONGO_DB` | Database name | `expense` |
-| `MONGO_EXPENSE_COLLECTION` | Expenses collection | `expenses` |
+| `MONGO_EXPENSE_COLLECTION` | Expenses collection | `transactions` |
 | `MONGO_ALIAS_COLLECTION` | Merchant alias collection | `merchant_aliases` |
 | `MONGO_SUGGESTION_COLLECTION` | Merge-suggestion collection | `merge_suggestions` |
 | `AMOUNT_TOLERANCE_PCT` | Amount delta percentage | `1.0` |
 | `TIME_TOLERANCE_MINUTES` | Time delta minutes | `10` |
 | `DEFAULT_LOOKBACK_HOURS` | How far back to scan | `48` |
+| `TIME_FIELDS` | CSV priority order for timestamps | `date,expense_time,createdAt` |
+| `MERCHANT_FIELDS` | CSV priority order for merchant labels | `merchant,note,paymentType,type,to` |
+| `USER_FIELD` | Source field that stores the user id (inferred automatically) | `user` |
+
+Smoke Test
+----------
+
+Use the bundled `test.py` script to hit the running API (locally or on the Hugging Face Space) via the base URL:
+
+```
+python3 test.py --base-url https://LogicGoInfotechSpaces-duplicate-transaction-detection.hf.space --lookback-hours 720 --limit 5000
+```
+
+The script calls `/health`, `/duplicates/detect`, and `/suggestions` in sequence and prints the responses so you can quickly verify the deployment.
 
 Next Steps
 ----------

requirements.txt

@@ -2,4 +2,5 @@ pymongo>=4.8.0
 python-dotenv>=1.0.1
 fastapi>=0.115.0
 uvicorn>=0.30.6
+requests>=2.32.3
 

src/api.py

@@ -33,8 +33,8 @@ suggestion_repository = MergeSuggestionRepository.from_client(mongo_client)
 class DetectRequest(BaseModel):
     lookback_hours: int | None = Field(
         default=None,
-        ge=1,
-        description="Hours to look back when scanning expenses.",
+        ge=0,
+        description="Hours to look back when scanning expenses (0 = full history).",
     )
     limit: int | None = Field(
         default=None,

src/config.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import os
 from dataclasses import dataclass
 from decimal import Decimal
+from typing import Tuple
 
 from dotenv import load_dotenv
 
@@ -31,11 +32,19 @@ def _get_int(env_key: str, default: str) -> int:
         raise ValueError(f"Invalid int for {env_key}: {raw_value}") from exc
 
 
+def _get_tuple(env_key: str, default: str) -> Tuple[str, ...]:
+    raw_value = os.getenv(env_key, default)
+    values = [value.strip() for value in raw_value.split(",") if value.strip()]
+    if not values:
+        raise ValueError(f"{env_key} must contain at least one value")
+    return tuple(values)
+
+
 @dataclass(frozen=True)
 class Settings:
     mongo_uri: str = os.getenv("MONGO_URI", DEFAULT_MONGO_URI)
     mongo_db: str = os.getenv("MONGO_DB", "expense")
-    expense_collection: str = os.getenv("MONGO_EXPENSE_COLLECTION", "expenses")
+    expense_collection: str = os.getenv("MONGO_EXPENSE_COLLECTION", "transactions")
     alias_collection: str = os.getenv("MONGO_ALIAS_COLLECTION", "merchant_aliases")
     suggestion_collection: str = os.getenv("MONGO_SUGGESTION_COLLECTION", "merge_suggestions")
     amount_tolerance_pct: Decimal = _get_decimal("AMOUNT_TOLERANCE_PCT", "1.0")
@@ -43,6 +52,14 @@ class Settings:
     default_lookback_hours: int = _get_int("DEFAULT_LOOKBACK_HOURS", "48")
     service_name: str = os.getenv("SERVICE_NAME", "duplicate-detector")
     max_batch_size: int = _get_int("MAX_BATCH_SIZE", "5000")
+    time_fields: Tuple[str, ...] = _get_tuple(
+        "TIME_FIELDS",
+        "date,expense_time,createdAt",
+    )
+    merchant_fields: Tuple[str, ...] = _get_tuple(
+        "MERCHANT_FIELDS",
+        "merchant,note,paymentType,type,to",
+    )
 
 
 settings = Settings()

src/duplicate_detector.py

@@ -45,6 +45,8 @@ class DuplicateDetector:
         for i, exp_a in enumerate(expenses):
             for j in range(i + 1, len(expenses)):
                 exp_b = expenses[j]
+                if exp_a.user_id and exp_b.user_id and exp_a.user_id != exp_b.user_id:
+                    continue
                 delta_minutes = _minutes_delta(exp_a.expense_time, exp_b.expense_time)
                 if delta_minutes > self.time_tolerance_minutes:
                     break

src/models.py

@@ -3,10 +3,32 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import datetime, timezone
 from decimal import Decimal
 from typing import List, Mapping, Sequence
 
+from .config import settings
+
+
+def _extract_expense_time(doc: Mapping[str, object]) -> datetime:
+    for field in settings.time_fields:
+        value = doc.get(field)
+        if isinstance(value, datetime):
+            if value.tzinfo is None:
+                value = value.replace(tzinfo=timezone.utc)
+            return value
+    raise ValueError("Expense document missing a valid timestamp field")
+
+
+def _extract_merchant(doc: Mapping[str, object]) -> str:
+    for field in settings.merchant_fields:
+        value = doc.get(field)
+        if value:
+            text = str(value).strip()
+            if text:
+                return text
+    return ""
+
 
 @dataclass(frozen=True)
 class Expense:
@@ -15,6 +37,7 @@ class Expense:
     currency: str
     merchant: str
     expense_time: datetime
+    user_id: str | None = None
     source: str | None = None
     metadata: Mapping[str, object] | None = None
 
@@ -24,12 +47,15 @@ class Expense:
             amount_value = Decimal(str(doc["amount"]))
         except KeyError as exc:
             raise ValueError("Expense document missing 'amount'") from exc
+        expense_time = _extract_expense_time(doc)
+        merchant_value = _extract_merchant(doc)
         return Expense(
             expense_id=str(doc.get("_id")),
             amount=amount_value,
             currency=str(doc.get("currency", "INR")),
-            merchant=str(doc.get("merchant", "")).strip(),
-            expense_time=doc["expense_time"],
+            merchant=merchant_value,
+            expense_time=expense_time,
+            user_id=str(doc.get("user")) if doc.get("user") else None,
             source=doc.get("source"),
             metadata=doc.get("metadata") or {},
         )

src/repositories.py

@@ -28,16 +28,21 @@ class ExpenseRepository:
         )
 
     def fetch_recent(self, lookback_hours: int, limit: int) -> List[Expense]:
-        since = datetime.now(tz=timezone.utc) - timedelta(hours=lookback_hours)
-        cursor = (
-            self._collection.find(
-                {"expense_time": {"$gte": since}},
-                sort=[("expense_time", 1)],
-                limit=limit,
-            )
-            or []
-        )
-        return [Expense.from_document(doc) for doc in cursor]
+        since: datetime | None = None
+        if lookback_hours > 0:
+            since = datetime.now(tz=timezone.utc) - timedelta(hours=lookback_hours)
+        if since:
+            time_filters = [
+                {field: {"$gte": since}}
+                for field in config.settings.time_fields
+            ]
+            query = {"$or": time_filters}
+        else:
+            query = {}
+        cursor = self._collection.find(query, limit=limit) or []
+        expenses = [Expense.from_document(doc) for doc in cursor]
+        expenses.sort(key=lambda expense: expense.expense_time)
+        return expenses
 
 
 class MerchantAliasRepository:

test.py

@@ -0,0 +1,71 @@
+"""Simple CLI to hit the duplicate-detector API."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from typing import Any, Dict
+
+import requests
+
+DEFAULT_BASE_URL = os.getenv("API_BASE_URL", "http://127.0.0.1:8000")
+
+
+def _print_response(label: str, response: requests.Response) -> None:
+    print(f"\n== {label} ==")
+    print(f"Status: {response.status_code}")
+    if response.headers.get("content-type", "").startswith("application/json"):
+        try:
+            payload = response.json()
+            print(json.dumps(payload, indent=2, default=str))
+        except ValueError:
+            print(response.text)
+    else:
+        print(response.text)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Exercise the duplicate-detector API.")
+    parser.add_argument(
+        "--base-url",
+        default=DEFAULT_BASE_URL,
+        help=f"API base URL (default: {DEFAULT_BASE_URL}).",
+    )
+    parser.add_argument("--lookback-hours", type=int, default=720)
+    parser.add_argument("--limit", type=int, default=5000)
+    parser.add_argument("--amount-pct", type=float, default=1.0)
+    parser.add_argument("--minutes", type=int, default=30)
+    args = parser.parse_args(argv)
+
+    base_url = args.base_url.rstrip("/")
+    payload: Dict[str, Any] = {
+        "lookback_hours": args.lookback_hours,
+        "limit": args.limit,
+        "amount_pct": args.amount_pct,
+        "minutes": args.minutes,
+    }
+
+    try:
+        health = requests.get(f"{base_url}/health", timeout=10)
+        _print_response("Health", health)
+
+        detect = requests.post(
+            f"{base_url}/duplicates/detect",
+            json=payload,
+            timeout=60,
+        )
+        _print_response("Detect", detect)
+
+        suggestions = requests.get(f"{base_url}/suggestions", timeout=10)
+        _print_response("Suggestions", suggestions)
+        return 0
+    except requests.RequestException as exc:
+        print(f"Request failed: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+