Fix/parse robustness

examples/django_reference/reference_app/ofr_bridge.py

@@ -146,10 +146,12 @@ def process_source_file(source_file_id: int) -> dict[str, Any]:
             normalized["transaction_date"] = pd.to_datetime(
                 normalized["transaction_date"]
             ).dt.date
-            normalized["units"] = normalized["units"].astype(float)
-            normalized["amount"] = normalized["amount"].astype(float)
+            from openreversefeed.adapters.parse_utils import to_number
+
+            normalized["units"] = normalized["units"].map(to_number)
+            normalized["amount"] = normalized["amount"].map(to_number)
             if "nav" in normalized.columns:
-                normalized["nav"] = normalized["nav"].astype(float)
+                normalized["nav"] = normalized["nav"].map(to_number)
 
             cleaned = Cleaner().run(normalized, adapter)
 

src/openreversefeed/adapters/cams.py

@@ -1,4 +1,5 @@
 """CAMS_FORMAT1 adapter. See spec §5 step 3."""
+
 from __future__ import annotations
 
 from pathlib import Path
@@ -32,6 +33,17 @@
     "REINVEST_F": "dividend_option_flag",
 }
 
+# CAMS WBR2C (DBF/CSV) uses SCHEME (name) + PRODCODE (code) instead of SCHEME_CODE.
+_WBR2C_FIELD_MAP: dict[str, str] = {
+    **_FIELD_MAP,
+    "TRXNNO": "transaction_number",
+    "PAN": "pan",
+    "INV_NAME": "investor_name",
+    "PURPRICE": "nav",
+    "STAMP_DUTY": "stamp_duty",
+    "AMC_CODE": "amc_code",
+}
+
 # Mapping from the raw CAMS REINVEST_F flag to the canonical plan_type
 # vocabulary the scheme master stores. 'Y' means reinvest, 'N' means
 # payout. Anything else (empty, 'P' for pure growth, nulls) → None,
@@ -192,12 +204,58 @@ def classify_row(self, row: dict[str, Any]) -> tuple[Action, str, bool]:
         return action, tag, False
 
     def composite_key(self, row: dict[str, Any]) -> str:
-        date_val = row["transaction_date"]
-        date_str = date_val.strftime("%Y%m%d") if hasattr(date_val, "strftime") else str(date_val)
+        from openreversefeed.core.composite_key import _date_str
+
         return (
             f"{row['original_trans_number']}_{row['transaction_type']}_"
-            f"{row['transaction_number']}_{date_str}"
+            f"{row['transaction_number']}_{_date_str(row['transaction_date'])}"
         )
 
 
+class CamsWbr2cAdapter(CamsAdapter):
+    """CAMS WBR2C transaction feed — DBF/CSV with SCHEME + AMC_CODE columns."""
+
+    name = "cams_wbr2c"
+    priority = 110
+    mandatory_headers = {
+        "USRTRXNO",
+        "FOLIO_NO",
+        "PRODCODE",
+        "SCHEME",
+        "UNITS",
+        "AMOUNT",
+        "TRADDATE",
+        "TRXNMODE",
+        "TRXNTYPE",
+    }
+    discriminator_headers = {"AMC_CODE", "TRXNNO"}
+    field_map = _WBR2C_FIELD_MAP
+
+    def parse(self, file_path: str | Path) -> pd.DataFrame:
+        path = Path(file_path)
+        if path.suffix.lower() == ".csv":
+            from openreversefeed.adapters.parse_utils import (
+                has_quoted_headers,
+                normalize_quoted_csv,
+            )
+
+            if has_quoted_headers(path):
+                return normalize_quoted_csv(path)
+        return super().parse(file_path)
+
+    def normalize(self, raw: pd.DataFrame) -> pd.DataFrame:
+        df = super().normalize(raw)
+        # WBR2C ships PRODCODE as the scheme identifier (no SCHEME_CODE column).
+        if "scheme_code" not in df.columns and "product_code" in df.columns:
+            df["scheme_code"] = df["product_code"]
+        # Include stamp duty in cost basis, matching Investwell / KFintech handling.
+        if "stamp_duty" in df.columns:
+            df["amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0) + pd.to_numeric(
+                df["stamp_duty"], errors="coerce"
+            ).fillna(0)
+            df.drop(columns=["stamp_duty"], inplace=True)
+        return df
+
+
 default_registry.register(CamsAdapter)
+default_registry.register(CamsWbr2cAdapter)

src/openreversefeed/adapters/kfintech.py

@@ -1,4 +1,5 @@
 """KFintech (formerly KARVY) adapters. Three sub-formats: FORMAT1, FORMAT2, CSV."""
+
 from __future__ import annotations
 
 from pathlib import Path
@@ -163,10 +164,10 @@ def classify_row(self, row: dict[str, Any]) -> tuple[Action, str, bool]:
         return action, tag, False
 
     def composite_key(self, row: dict[str, Any]) -> str:
+        from openreversefeed.core.composite_key import _date_str
+
         parent = row.get("parent_transaction_number") or "0"
-        date_val = row["transaction_date"]
-        date_str = date_val.strftime("%Y%m%d") if hasattr(date_val, "strftime") else str(date_val)
-        return f"{row['transaction_number']}_{parent}_{row['folio_number']}_{date_str}"
+        return f"{row['transaction_number']}_{parent}_{row['folio_number']}_{_date_str(row['transaction_date'])}"
 
 
 class KFintechFormat1Adapter(_KFintechBase):
@@ -237,6 +238,69 @@ def normalize(self, raw: pd.DataFrame) -> pd.DataFrame:
         return self._normalize_with_map(raw, _CSV_FIELD_MAP)
 
 
+# Field map for the KFintech "detailed" English-header CSV format that broker
+# portals / BSE STAR MF export. Column names differ from the simple CSV format
+# in capitalization and naming (e.g. "Nav" not "NAV", "Transaction Head" not
+# "Transaction Purred", "PAN1" not "PAN", "Product Code" not "AMC Code").
+_DETAILED_CSV_FIELD_MAP = {
+    "Inward Number": "transaction_id",
+    "Transaction Number": "transaction_number",
+    "Purchase Transaction No": "parent_transaction_number",
+    "Folio Number": "folio_number",
+    "Product Code": "product_code",
+    "Scheme Code": "scheme_code",
+    "Units": "units",
+    "Amount": "amount",
+    "Nav": "nav",
+    "Transaction Date": "transaction_date",
+    "Transaction Mode": "transaction_mode",
+    # "Transaction Head" carries P/R/D/DP — the purred-style indicator
+    "Transaction Head": "transaction_purred",
+    # "Transaction Flag" carries TI/TO/SI/SO for transfers and switches
+    "Transaction Flag": "transaction_flag",
+    "Transaction Type": "transaction_type",
+    "PAN1": "pan",
+    "Investor Name": "investor_name",
+    "Agent Code": "broker_code",
+    # Stamp duty is a separate column; we capture it so normalize() can fold
+    # it into the canonical amount for correct cost-basis accounting.
+    "Stamp Duty Charges": "stamp_duty",
+}
+
+
+class KFintechDetailedCsvAdapter(_KFintechBase):
+    """KFintech English-header CSV as exported by BSE STAR MF / broker portals.
+
+    Distinguishable from KFintechCsvAdapter by:
+    - "Nav" (not "NAV") as the NAV column
+    - "Transaction Head" (not "Transaction Purred") for the P/R/D indicator
+    - "PAN1" (not "PAN") for the investor PAN
+    - "Product Code" (not "AMC Code") for the AMC identifier
+    """
+
+    name = "kfintech_detailed_csv"
+    priority = 75
+    mandatory_headers = {"Inward Number", "Folio Number", "Units", "Transaction Date"}
+    discriminator_headers = {"Transaction Head", "PAN1"}
+    field_map = _DETAILED_CSV_FIELD_MAP
+
+    def parse(self, file_path: str | Path) -> pd.DataFrame:
+        return pd.read_csv(Path(file_path), dtype=str)
+
+    def normalize(self, raw: pd.DataFrame) -> pd.DataFrame:
+        df = self._normalize_with_map(raw, _DETAILED_CSV_FIELD_MAP)
+        # Fold stamp duty into amount so cost-basis reflects the true debit.
+        # Stamp duty is always a cost on purchases; on redemptions it is zero.
+        if "stamp_duty" in df.columns:
+            df["amount"] = (
+                pd.to_numeric(df["amount"], errors="coerce").fillna(0)
+                + pd.to_numeric(df["stamp_duty"], errors="coerce").fillna(0)
+            ).astype(str)
+            df.drop(columns=["stamp_duty"], inplace=True)
+        return df
+
+
 default_registry.register(KFintechFormat1Adapter)
 default_registry.register(KFintechFormat2Adapter)
 default_registry.register(KFintechCsvAdapter)
+default_registry.register(KFintechDetailedCsvAdapter)

src/openreversefeed/adapters/parse_utils.py

@@ -0,0 +1,71 @@
+"""Feed parsing helpers — quote normalization, header peeking, numeric coercion."""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+
+def _strip_outer_quotes(value: str) -> str:
+    """Strip matched leading/trailing single quotes from a string value."""
+    if len(value) >= 2 and value[0] == "'" and value[-1] == "'":
+        return value[1:-1]
+    return value
+
+
+def normalize_quoted_csv(file_path: str | Path) -> pd.DataFrame:
+    """Read a CSV that may have single-quoted headers and cell values.
+
+    Strips matched leading/trailing single quotes from column names and
+    from all string cell values. Leaves unquoted values and internal
+    apostrophes untouched.
+    """
+    df = pd.read_csv(Path(file_path), dtype=str)
+    df.columns = pd.Index([_strip_outer_quotes(c) for c in df.columns])
+    for col in df.columns:
+        if df[col].dtype == object:
+            df[col] = df[col].map(lambda v: _strip_outer_quotes(v) if isinstance(v, str) else v)
+    return df
+
+
+def peek_feed_headers(file_path: str | Path) -> set[str]:
+    """Read only the first line of a CSV and return normalized header names.
+
+    Applies the same single-quote stripping as normalize_quoted_csv so
+    detection works on quoted CAMS lot-level files.
+    """
+    path = Path(file_path)
+    with path.open(newline="", encoding="utf-8-sig") as f:
+        reader = csv.reader(f)
+        try:
+            raw_headers = next(reader)
+        except StopIteration:
+            return set()
+    return {_strip_outer_quotes(h.strip()) for h in raw_headers if h.strip()}
+
+
+def has_quoted_headers(file_path: str | Path) -> bool:
+    """Return True if the CSV header row contains single-quoted column names."""
+    path = Path(file_path)
+    with path.open(newline="", encoding="utf-8-sig") as f:
+        first_line = f.readline()
+    return first_line.startswith("'") and "'," in first_line
+
+
+def to_number(value: Any) -> float:
+    """Convert a value to float, stripping commas from Indian-formatted numbers.
+
+    Returns float('nan') for empty strings and None.
+    """
+    if isinstance(value, (int, float)):
+        return float(value)
+    if value is None:
+        return float("nan")
+    s = str(value).strip()
+    if not s:
+        return float("nan")
+    s = s.replace(",", "")
+    return float(s)

src/openreversefeed/core/composite_key.py

@@ -1,17 +1,84 @@
 """Deterministic composite key builders. See spec §5 step 5."""
+
 from __future__ import annotations
 
+import re
 from typing import Any
 
 import pandas as pd
 
 from openreversefeed.core.models import Registrar
 
+_JS_DATE_RE = re.compile(r"[A-Z][a-z]{2}\s+([A-Z][a-z]{2})\s+(\d{1,2})\s+(\d{4})")
+_MONTH_ABBR = {
+    "Jan": 1,
+    "Feb": 2,
+    "Mar": 3,
+    "Apr": 4,
+    "May": 5,
+    "Jun": 6,
+    "Jul": 7,
+    "Aug": 8,
+    "Sep": 9,
+    "Oct": 10,
+    "Nov": 11,
+    "Dec": 12,
+}
+_SLASH_AMPM_RE = re.compile(
+    r"^(\d{1,2})/(\d{1,2})/(\d{4})\s+\d{1,2}:\d{2}:\d{2}\s+[AP]M$",
+    re.IGNORECASE,
+)
+_ISO_RE = re.compile(r"^(\d{4})-(\d{2})-(\d{2})$")
+_SLASH_RE = re.compile(r"^(\d{1,2})/(\d{1,2})/(\d{4})$")
+_DASH_DMY_RE = re.compile(r"^(\d{1,2})-(\d{1,2})-(\d{4})$")
+_ALREADY_YYYYMMDD_RE = re.compile(r"^\d{8}$")
+
 
 def _date_str(value: Any) -> str:
     if hasattr(value, "strftime"):
         return value.strftime("%Y%m%d")
-    return str(value)
+
+    s = str(value).strip()
+
+    if _ALREADY_YYYYMMDD_RE.match(s):
+        return s
+
+    # ISO: YYYY-MM-DD
+    m = _ISO_RE.match(s)
+    if m:
+        return f"{m.group(1)}{m.group(2)}{m.group(3)}"
+
+    # US-style with AM/PM timestamp: M/D/YYYY HH:MM:SS AM
+    m = _SLASH_AMPM_RE.match(s)
+    if m:
+        month, day, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
+        return f"{year:04d}{month:02d}{day:02d}"
+
+    # Slash without timestamp: first > 12 means DD/MM, else DD/MM (Indian default)
+    m = _SLASH_RE.match(s)
+    if m:
+        a, b, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
+        if a > 12:
+            day, month = a, b
+        else:
+            day, month = a, b
+        return f"{year:04d}{month:02d}{day:02d}"
+
+    # Dash DD-MM-YYYY
+    m = _DASH_DMY_RE.match(s)
+    if m:
+        day, month, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
+        return f"{year:04d}{month:02d}{day:02d}"
+
+    # JavaScript Date.toString(): "Fri Aug 28 2026 ..."
+    m = _JS_DATE_RE.match(s)
+    if m:
+        month_abbr, day_s, year_s = m.group(1), m.group(2), m.group(3)
+        month = _MONTH_ABBR.get(month_abbr, 0)
+        if month:
+            return f"{int(year_s):04d}{month:02d}{int(day_s):02d}"
+
+    return s
 
 
 def build_cams_key(row: dict[str, Any]) -> str: