Add Python reconciliation engine

recon/engine.py

@@ -0,0 +1,213 @@
+"""
+Addepar <-> iCapital reconciliation engine for alternatives and insurance products.
+
+iCapital is treated as the book of record for subscription, capital activity and
+NAV/contract value. Addepar is the reporting system that has to match it. The
+engine:
+
+  1. maps every iCapital holding to its Addepar security and position,
+  2. reconciles commitments, capital activity (calls, distributions, premiums)
+     and valuations,
+  3. auto-resolves the breaks that follow a deterministic rule (unbooked events,
+     misclassified distributions, valuation lag, rounding) and writes them to an
+     Addepar import file, and
+  4. routes only true exceptions to a person, each with a suggested fix.
+"""
+
+from __future__ import annotations
+
+import csv
+from collections import defaultdict
+from dataclasses import asdict, dataclass
+from datetime import date, timedelta
+from pathlib import Path
+
+ROUNDING_TOLERANCE = 1.00          # dollars
+STALE_AFTER_DAYS = 120             # valuation age that needs a fresh GP/sponsor statement
+INSURANCE = {"PPLI", "PPVA", "Fixed Indexed Annuity", "RILA"}
+
+# Effort model used for the automation metrics (minutes).
+MANUAL_REVIEW_PER_POSITION = 12    # rep ties out a position by hand
+MANUAL_FIX_PER_BREAK = 20          # rep researches and books a fix
+EXCEPTION_REVIEW = 15              # analyst works an exception the engine couldn't clear
+
+
+@dataclass
+class Finding:
+    account_id: str
+    household: str
+    security: str
+    asset_class: str
+    check: str
+    status: str            # Matched | Auto-resolved | Exception
+    icapital: float | None
+    addepar: float | None
+    difference: float | None
+    detail: str
+    action: str
+
+
+def _read(path: Path) -> list[dict]:
+    with open(path, newline="") as f:
+        return list(csv.DictReader(f))
+
+
+def load(data_dir: str | Path) -> dict:
+    d = Path(data_dir)
+    return {n: _read(d / f"{n}.csv") for n in
+            ("security_master", "accounts", "icapital_positions", "icapital_activity",
+             "addepar_positions", "addepar_transactions")}
+
+
+def reconcile(data: dict, as_of: date = date(2026, 9, 30)) -> dict:
+    sec_by_ic = {s["icapital_id"]: s for s in data["security_master"]}
+    household = {a["account_id"]: a["household"] for a in data["accounts"]}
+    add_pos = {(p["account_id"], p["addepar_id"]): p for p in data["addepar_positions"]}
+    events = defaultdict(list)
+    for e in data["icapital_activity"]:
+        events[(e["account_id"], e["icapital_id"])].append(e)
+    txns = defaultdict(list)
+    for t in data["addepar_transactions"]:
+        txns[(t["account_id"], t["addepar_id"])].append(t)
+
+    findings: list[Finding] = []
+    imports: list[dict] = []
+
+    for ip in data["icapital_positions"]:
+        acct, ic = ip["account_id"], ip["icapital_id"]
+        sec = sec_by_ic.get(ic, {"name": ic, "asset_class": "Unknown", "addepar_id": ""})
+        cls, name, ad = sec["asset_class"], sec["name"], sec["addepar_id"]
+        nav, called, commit = float(ip["nav"]), float(ip["called_to_date"]), float(ip["commitment"])
+        ev = events[(acct, ic)]
+
+        def add(check, status, icv=None, adv=None, detail="", action=""):
+            diff = None if icv is None or adv is None else round(adv - icv, 2)
+            findings.append(Finding(acct, household.get(acct, acct), name, cls, check, status,
+                                    icv, adv, diff, detail, action))
+
+        # --- Mapping and position existence -----------------------------------
+        if not ad:
+            add("Security mapping", "Exception", nav, None,
+                f"{name} exists in iCapital but has no Addepar security.",
+                f"Create the Addepar security, map {ic}, then the engine will book "
+                f"{len(ev)} activity event(s) and the current NAV on the next run.")
+            continue
+        ap = add_pos.get((acct, ad))
+        if ap is None:
+            add("Position setup", "Exception", nav, None,
+                f"{acct} holds {name} in iCapital (${commit:,.0f} commitment) but has no Addepar position.",
+                "Open the position in Addepar with the iCapital commitment; activity and NAV will "
+                "flow automatically on the next run.")
+            continue
+
+        count_before = len(findings)
+
+        # --- Commitment ------------------------------------------------------
+        a_commit = float(ap["commitment"])
+        if abs(a_commit - commit) > ROUNDING_TOLERANCE:
+            add("Commitment", "Exception", commit, a_commit,
+                "Commitment in Addepar does not match the subscription on iCapital.",
+                "Verify against the signed subscription document, then correct the Addepar commitment "
+                "(it drives unfunded commitment and pacing reports).")
+
+        # --- Capital activity ------------------------------------------------
+        by_ref = defaultdict(list)
+        for t in txns[(acct, ad)]:
+            by_ref[t["source_ref"]].append(t)
+        for ref, group in by_ref.items():
+            if len(group) > 1:
+                amt = float(group[0]["amount"])
+                add("Duplicate transaction", "Exception", amt, amt * len(group),
+                    f"{group[0]['type']} for ${amt:,.2f} ({ref}) is booked {len(group)} times in Addepar.",
+                    f"Approve removal of {', '.join(t['txn_id'] for t in group[1:])}; deletes need a human sign-off.")
+        for e in ev:
+            amt = float(e["amount"])
+            booked = by_ref.get(e["event_id"])
+            if not booked:
+                add(f"{e['type']} not booked", "Auto-resolved", amt, 0.0,
+                    f"{e['type']} of ${amt:,.2f} (notice {e['notice_date']}) is on iCapital but not in Addepar.",
+                    f"Booked {e['type'].lower()} dated {e['due_date']} to Addepar import.")
+                imports.append({"action": "ADD_TRANSACTION", "account_id": acct, "addepar_id": ad,
+                                "type": e["type"], "amount": amt, "date": e["due_date"], "source_ref": e["event_id"]})
+            elif booked[0]["type"] != e["type"]:
+                add("Distribution classification", "Auto-resolved", amt, amt,
+                    f"{e['event_id']} is {e['type']} on iCapital but booked as {booked[0]['type']} in Addepar.",
+                    "Reclassified in Addepar import; cost basis and income reporting corrected.")
+                imports.append({"action": "RECLASSIFY", "account_id": acct, "addepar_id": ad,
+                                "type": e["type"], "amount": amt, "date": booked[0]["trade_date"],
+                                "source_ref": e["event_id"]})
+
+        # --- Valuation -------------------------------------------------------
+        nav_date = date.fromisoformat(ip["nav_date"])
+        a_val, a_date = float(ap["market_value"]), date.fromisoformat(ap["valuation_date"])
+        label = "Contract value" if cls in INSURANCE else "NAV"
+        source = "carrier statement" if cls in INSURANCE else "sponsor statement"
+        if (as_of - nav_date).days > STALE_AFTER_DAYS:
+            add("Stale valuation", "Exception", nav, a_val,
+                f"Latest {label} is dated {nav_date.isoformat()} ({(as_of - nav_date).days} days old).",
+                f"Request the current {source}; performance for this position is understated or overstated until it posts.")
+        elif a_date < nav_date:
+            add(f"{label} lag", "Auto-resolved", nav, a_val,
+                f"Addepar is on the {a_date.isoformat()} mark; iCapital has {nav_date.isoformat()}.",
+                f"Posted {label} of ${nav:,.2f} as of {nav_date.isoformat()} to Addepar import.")
+            imports.append({"action": "UPDATE_VALUATION", "account_id": acct, "addepar_id": ad,
+                            "type": label, "amount": nav, "date": nav_date.isoformat(), "source_ref": ic})
+        elif abs(a_val - nav) <= ROUNDING_TOLERANCE and a_val != nav:
+            add("Rounding", "Auto-resolved", nav, a_val, "Difference is within the $1.00 rounding tolerance.",
+                "Cleared automatically; no booking needed.")
+        elif abs(a_val - nav) > ROUNDING_TOLERANCE:
+            add(f"{label} break", "Exception", nav, a_val,
+                f"Same valuation date ({nav_date.isoformat()}) but values differ by "
+                f"{abs(a_val - nav) / nav:.1%}.",
+                f"Tie out to the {source}; if iCapital is correct, approve the Addepar update.")
+
+        if len(findings) == count_before:
+            add("All checks", "Matched", nav, a_val, "Commitment, activity and valuation agree.", "")
+
+    return {"findings": findings, "imports": imports, "metrics": metrics(findings, data)}
+
+
+def metrics(findings: list[Finding], data: dict) -> dict:
+    positions = len(data["icapital_positions"])
+    auto = sum(f.status == "Auto-resolved" for f in findings)
+    exc = sum(f.status == "Exception" for f in findings)
+    by_pos = defaultdict(set)
+    for f in findings:
+        by_pos[(f.account_id, f.security)].add(f.status)
+    clean = sum(s == {"Matched"} for s in by_pos.values())
+    manual_min = positions * MANUAL_REVIEW_PER_POSITION + (auto + exc) * MANUAL_FIX_PER_BREAK
+    auto_min = exc * EXCEPTION_REVIEW
+    return {
+        "positions": positions,
+        "clean_positions": clean,
+        "breaks": auto + exc,
+        "auto_resolved": auto,
+        "exceptions": exc,
+        "auto_resolution_rate": round(auto / (auto + exc), 3) if auto + exc else 1.0,
+        "manual_hours": round(manual_min / 60, 1),
+        "automated_hours": round(auto_min / 60, 1),
+        "hours_saved": round((manual_min - auto_min) / 60, 1),
+        "positions_touched_by_people": len({(f.account_id, f.security) for f in findings if f.status == "Exception"}),
+    }
+
+
+def write_outputs(result: dict, out_dir: str | Path) -> None:
+    out = Path(out_dir)
+    out.mkdir(parents=True, exist_ok=True)
+    rows = [asdict(f) for f in result["findings"]]
+    with open(out / "recon_findings.csv", "w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
+        w.writeheader()
+        w.writerows(rows)
+    with open(out / "addepar_import.csv", "w", newline="") as f:
+        w = csv.DictWriter(f, fieldnames=["action", "account_id", "addepar_id", "type", "amount", "date", "source_ref"])
+        w.writeheader()
+        w.writerows(result["imports"])
+
+
+if __name__ == "__main__":
+    here = Path(__file__).resolve().parent.parent
+    res = reconcile(load(here / "data"))
+    write_outputs(res, here / "output")
+    for k, v in res["metrics"].items():
+        print(f"{k:28s} {v}")