feat(policies): capture the full premium breakdown

The capture form only ever had prima neta, derecho de póliza and comisión.
The Access form it replaces has seven figures, and the four that were missing
are the ones that make a policy paid in installments add up.

Adds recargo, IVA, prima total and forma de pago to the policy header, the
same breakdown per installment, and a per-line-of-business IVA rate.

IVA and prima total are the only derived figures:

    base  = prima neta + recargo + derecho de póliza
    IVA   = round(base * tasa)
    total = base + IVA

The recargo is inside the taxable base. That is not a guess — policy 7006785
prints IVA 52.03 on 610.86 + 8.55 + 31.00, and leaving the recargo out gives
51.35, which matches nothing on the page. Both of its money rows are asserted
in premium.spec.ts. The recargo itself is never derived: the carrier quotes it,
so staff key it in, and the field is disabled on ANNUAL/SINGLE. Both derived
figures are stored rather than recomputed on read, and stay editable, because
the printed policy is the record of truth and a later rate change must not
silently restate what was issued.

The rate lives on PolicyType (seeded to 0.08, editable in Catálogos), which is
the legacy one-row IMPUESTOS / IMPUESTOS_AUTOS tables made configurable. The
rate applied is stamped on the policy so an old one reads back at its original
rate.

Per-installment, not two fixed slots on the header: a policy split into several
exhibiciones prices each payment separately — that is why the Access form drew
the money row twice — and a trimestral policy needs four, which the Access
layout could not hold.

Also fixes two losses in the ETL, which is how these went missing:

  - `forma_pago` was marked consumed by the coverage sweep and then never
    written to any column, so FORMA PAGO existed nowhere in the platform.
  - `recargo` and the whole second money row fell into `coveragesJson` as
    loose strings, mislabeled as coverage amounts.

transform_policies.py now writes all of it directly;
backfill_policy_premium_breakdown.py recovers it on a database that must not be
re-imported, and strips the migrated keys back out of coveragesJson. Both are
COALESCE-only, so a figure a human has corrected in the app wins.

IVA and TOTAL are NOT backfilled: they were unbound calculated controls on the
Access form, never columns, so there is nothing to recover and every migrated
policy reads null until it is edited.

The backfill warns on 5 annual policies that carry a non-zero recargo — a
contradiction that predates this change and is left for a human, not silently
corrected.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

migration/backfill_policy_premium_breakdown.py

@@ -0,0 +1,249 @@
+"""
+Recovers the premium breakdown the original policy transform dropped.
+
+`transform_policies.py` modeled prima neta, derecho de póliza and comisión and
+nothing else, which lost two things from every migrated policy:
+
+  1. RECARGO — the financing surcharge on a policy paid in more than one
+     exhibición, plus the whole second money row (`p_neta_2`, `recargo_2`,
+     `d_pol_2`, `com_2`) that a semestral policy carries because each payment
+     is priced separately. These were not deleted, they were swept into
+     `policies.coveragesJson` as loose strings alongside the real coverages —
+     unqueryable, and mislabeled as coverage amounts.
+
+  2. FORMA PAGO — dropped outright. The column was marked "consumed" by the
+     transform's coverage sweep but never written to any column, so it exists
+     nowhere in the platform database. It is the field that decides whether a
+     recargo is legitimate on a row at all, so it cannot be inferred back from
+     the money.
+
+The transform has been fixed in the same commit, so a full `run_all.py` now
+produces all of this directly. This script exists for a database that must not
+be re-imported: it reads the same staged Parquet and patches in place.
+
+What it does NOT do: invent IVA or the printed TOTAL. Those were never columns
+in the home tables — they were unbound calculated controls on the Access form —
+so there is genuinely nothing to recover, and both stay null until a human
+edits the policy. The app computes them (apps/api/src/policies/premium.ts).
+
+Idempotent, and never overwrites a non-null value: a figure a human has since
+corrected in the app wins over the legacy one.
+
+    ./.venv/bin/python backfill_policy_premium_breakdown.py --env dev
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from decimal import Decimal, InvalidOperation
+from pathlib import Path
+
+import pandas as pd
+
+from dbenv import connect
+from sync import parse_mode
+
+STG = Path(__file__).parent / "output" / "stg_seguros"
+LEGACY_DB = "SEGUROS 16_be"
+NULL = "∅"
+
+# Legacy column -> what it is, per source table. Only the tables that actually
+# carry a breakdown appear; the auto tables have no recargo and no second row.
+HOME_TABLES = ("mult", "incendio", "m_empr")
+
+# Keys the transform used to dump into coveragesJson that are now real columns.
+# Stripped once migrated so the blob stops pretending they are coverages.
+MIGRATED_COVERAGE_KEYS = (
+    "recargo", "recargo_2", "p_neta_2", "d_pol_2", "com_2",
+)
+
+_FREQ = {
+    "ANNUAL": "ANNUAL",
+    "ANUAL": "ANNUAL",
+    "SEMESTRAL": "SEMIANNUAL",
+    "TRIMESTRAL": "QUARTERLY",
+    "MENSUAL": "MONTHLY",
+    "CONTADO": "SINGLE",
+}
+
+
+def s(v):
+    if v is None or pd.isna(v):
+        return None
+    v = str(v).strip()
+    return None if v in ("", NULL) else v
+
+
+def dec(v):
+    v = s(v)
+    if v is None:
+        return None
+    try:
+        return Decimal(v.replace(",", ""))
+    except (InvalidOperation, ValueError):
+        return None
+
+
+def freq(v):
+    return _FREQ.get((s(v) or "").upper())
+
+
+def load(name):
+    df = pd.read_parquet(STG / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True)
+    for c in df.columns:
+        if c != "_row_num":
+            df[c] = df[c].astype("string").str.strip()
+    return df
+
+
+def main():
+    env, _sync_mode = parse_mode()
+
+    # Fails closed rather than reporting a clean run over nothing: an empty
+    # staging directory and a policy set with no recargo look identical from
+    # the database side, and "0 rows updated" would read as success.
+    if not STG.exists():
+        print(f"[policy-premium] staged Parquet missing at {STG} — run extract/load first.")
+        return 3
+
+    conn = connect(env)
+    c = conn.cursor()
+    print(f"[policy-premium] target env: {env}")
+
+    # Every policy that came from the insurance ETL, keyed by provenance. The
+    # id is needed to reach the installments, coveragesJson to strip the keys.
+    c.execute(
+        "SELECT legacySourceTable, legacyId, id, coveragesJson "
+        "FROM policies WHERE legacySourceDb = %s AND legacyId IS NOT NULL",
+        (LEGACY_DB,),
+    )
+    by_key = {(t, lid): (pid, cov) for t, lid, pid, cov in c.fetchall()}
+    print(f"  {len(by_key)} migrated polic(ies) in the target database")
+
+    pol_updates = []       # (surcharge, paymentFrequency, coveragesJson, policyId)
+    inst_updates = []      # (netPremium, surcharge, policyFee, commission, policyId, seq)
+    seen_tables = 0
+
+    for table in HOME_TABLES + (
+        "tabla_autos", "tabla_autos_ampl", "tabla_autos_limit",
+        "tabla_autos_ampl_r", "tabla_autos_rc_r", "mca2", "licencias",
+    ):
+        path = STG / f"{table}.parquet"
+        if not path.exists():
+            continue
+        seen_tables += 1
+        df = load(table)
+        home = table in HOME_TABLES
+
+        for _, row in df.iterrows():
+            key = (table, str(int(row["_row_num"])))
+            hit = by_key.get(key)
+            if not hit:
+                continue
+            pid, cov_raw = hit
+
+            surcharge = dec(row.get("recargo")) if home else None
+            frequency = freq(row.get("forma_pago"))
+
+            # Strip the now-modeled keys out of the coverage blob. Rewritten
+            # only when something actually changes, so a policy whose blob a
+            # human has edited is left byte-identical.
+            cov_new = None
+            if cov_raw:
+                try:
+                    cov = json.loads(cov_raw) if isinstance(cov_raw, str) else cov_raw
+                except (TypeError, ValueError):
+                    cov = None
+                if isinstance(cov, dict):
+                    kept = {k: v for k, v in cov.items() if k not in MIGRATED_COVERAGE_KEYS}
+                    if len(kept) != len(cov):
+                        cov_new = json.dumps(kept, ensure_ascii=False) if kept else None
+
+            if surcharge is not None or frequency is not None or cov_new is not None:
+                pol_updates.append((surcharge, frequency, cov_new, cov_new is not None, pid))
+
+            # Per-payment breakdown. Slot 1 is the unsuffixed money row, slot 2
+            # the _2 twin; the auto tables have a single slot and no recargo.
+            if home:
+                slots = [
+                    (1, "p_neta", "recargo", "d_pol", "com"),
+                    (2, "p_neta_2", "recargo_2", "d_pol_2", "com_2"),
+                ]
+            else:
+                pn = "prima1" if table == "mca2" else "prima_neta"
+                dp = "d_poliza1" if table == "mca2" else "d_poliza"
+                slots = [(1, pn, None, dp, None)]
+
+            for seq, pn, rc, dp, cm in slots:
+                vals = (
+                    dec(row.get(pn)) if pn else None,
+                    dec(row.get(rc)) if rc else None,
+                    dec(row.get(dp)) if dp else None,
+                    dec(row.get(cm)) if cm else None,
+                )
+                if all(v is None for v in vals):
+                    continue
+                inst_updates.append((*vals, pid, seq))
+
+    if not seen_tables:
+        print(f"[policy-premium] no policy tables staged under {STG} — nothing to do.")
+        return 3
+
+    # COALESCE on every target: a column a human has already filled in the app
+    # keeps its value, the legacy figure only lands where there is a hole.
+    for surcharge, frequency, cov_new, rewrite_cov, pid in pol_updates:
+        c.execute(
+            "UPDATE policies SET "
+            "  surcharge = COALESCE(surcharge, %s), "
+            "  paymentFrequency = COALESCE(paymentFrequency, %s), "
+            "  coveragesJson = IF(%s, %s, coveragesJson) "
+            "WHERE id = %s",
+            (surcharge, frequency, 1 if rewrite_cov else 0, cov_new, pid),
+        )
+
+    for netp, surch, fee, comm, pid, seq in inst_updates:
+        c.execute(
+            "UPDATE policy_payment_installments SET "
+            "  netPremium = COALESCE(netPremium, %s), "
+            "  surcharge  = COALESCE(surcharge, %s), "
+            "  policyFee  = COALESCE(policyFee, %s), "
+            "  commission = COALESCE(commission, %s) "
+            "WHERE policyId = %s AND sequence = %s",
+            (netp, surch, fee, comm, pid, seq),
+        )
+
+    conn.commit()
+    print(f"  policies    : {len(pol_updates)} row(s) touched")
+    print(f"  installments: {len(inst_updates)} row(s) touched")
+
+    # --- validation ---------------------------------------------------------
+    c.execute("SELECT COUNT(*) FROM policies WHERE surcharge IS NOT NULL AND surcharge <> 0")
+    n_surch = c.fetchone()[0]
+    c.execute("SELECT COUNT(*) FROM policies WHERE paymentFrequency IS NOT NULL")
+    n_freq = c.fetchone()[0]
+    c.execute(
+        "SELECT COUNT(*) FROM policies "
+        "WHERE paymentFrequency IN ('ANNUAL','SINGLE') AND surcharge IS NOT NULL AND surcharge <> 0"
+    )
+    n_bad = c.fetchone()[0]
+    c.execute(
+        "SELECT COUNT(*) FROM policies WHERE coveragesJson IS NOT NULL "
+        "AND JSON_EXTRACT(coveragesJson, '$.recargo') IS NOT NULL"
+    )
+    n_left = c.fetchone()[0]
+
+    print(f"  -> policies with a recargo    : {n_surch}")
+    print(f"  -> policies with a forma pago : {n_freq}")
+    print(f"  -> recargo still in coverages : {n_left}")
+
+    # A surcharge on an annual policy contradicts the rule the capture form
+    # enforces, so it is worth surfacing rather than leaving for someone to
+    # find in a total. It is a warning, not a failure: the books are the books.
+    if n_bad:
+        print(f"  !! {n_bad} annual/contado polic(ies) carry a non-zero recargo — review by hand")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main() or 0)

migration/run_all.py

@@ -59,6 +59,12 @@ STEPS = [
     # touches.
     "backfill_statement_match_fields.py",
     "transform_policies.py",
+    # Premium breakdown (recargo, per-payment figures, forma de pago).
+    # transform_policies.py now writes these directly, so on a full rebuild
+    # this is a no-op that re-asserts them; on a database migrated before the
+    # breakdown existed it is what recovers them out of coveragesJson.
+    # Must follow transform_policies.py, which truncates the installments.
+    "backfill_policy_premium_breakdown.py",
     "transform_transactions.py",
     "prune_empty_customers.py",
     # Seeds the Scotiabank chequera that every SCOTHIA movement is booked into;
@@ -78,6 +84,12 @@ SYNC_STEPS = [
     # touches.
     "backfill_statement_match_fields.py",
     "transform_policies.py",
+    # Premium breakdown (recargo, per-payment figures, forma de pago).
+    # transform_policies.py now writes these directly, so on a full rebuild
+    # this is a no-op that re-asserts them; on a database migrated before the
+    # breakdown existed it is what recovers them out of coveragesJson.
+    # Must follow transform_policies.py, which truncates the installments.
+    "backfill_policy_premium_breakdown.py",
     "transform_transactions.py",
     # Manual-safe prune: drops legacy-owned empties that the customer upsert
     # re-creates from Parquet, but leaves manually-added customers alone.

migration/transform_policies.py

@@ -18,6 +18,17 @@ Design (validated against staged data):
     policies are skipped and counted (required FK).
   - Payment slots: c_1er_pago is the first amount, pago_subsec the recurring
     amount for slots 2-4; efectivo is a cash flag, no_cheque the check ref.
+  - Premium breakdown: a policy paid in more than one exhibicion prices EACH
+    payment separately, which is why the home tables carry the whole money row
+    twice (p_neta/recargo/d_pol/com and their _2 twins). The unsuffixed set is
+    the policy header, the suffixed one belongs to payment 2, and both are
+    written per installment as well. This used to be lost: `recargo` and the
+    _2 columns fell into coveragesJson as loose strings and forma_pago was
+    marked consumed but never written anywhere at all.
+  - IVA and the printed TOTAL are NOT in Access for the home tables. They were
+    unbound calculated controls on the form, so there is nothing to migrate;
+    the app computes them (apps/api/src/policies/premium.ts) from
+    (p_neta + recargo + d_pol) * rate.
   - Any source column not explicitly modeled (coverage amounts: edificio,
     contenidos, robo, cristales, ...) is preserved verbatim in coveragesJson,
     so nothing is lost in consolidation.
@@ -93,20 +104,55 @@ def truthy(v):
     return (s(v) or "0").lower() in {"1", "-1", "true", "si", "sí", "yes", "x"}
 
 
+# Access FORMA PAGO -> PaymentFrequency. The whole staged corpus holds exactly
+# four spellings (ANNUAL 1851, SEMESTRAL 29, semestral 2, CONTADO 1); anything
+# else is left null rather than guessed, because the value decides whether a
+# recargo is legitimate on the row.
+_FREQ = {
+    "ANNUAL": "ANNUAL",
+    "ANUAL": "ANNUAL",
+    "SEMESTRAL": "SEMIANNUAL",
+    "TRIMESTRAL": "QUARTERLY",
+    "MENSUAL": "MONTHLY",
+    "CONTADO": "SINGLE",
+}
+
+
+def freq(v):
+    return _FREQ.get((s(v) or "").upper())
+
+
+def slot_dec(row, slot, key):
+    """One figure of a payment's premium breakdown, or None when the source
+    table has no such column. Deliberately not zero: a zero d_pol on the second
+    payment is a real figure in the books and must stay distinguishable from
+    'this table never had that column'."""
+    col = slot.get(key)
+    return dec(row.get(col)) if col else None
+
+
 # --- per-table config ------------------------------------------------------- #
 # fields: policy column -> source column. installments: list of slot dicts.
 # vehicles: 'trip_underscore' | 'single' | 'mca2' | None. drivers: 'mca2' |
 # 'licencias' | None.
+# `pneta`/`recarg`/`dpol`/`com` on a slot are that payment's own share of the
+# premium. Only the first two payments have one in Access — the form only ever
+# drew the money row twice — so slots 3 and 4 carry none and keep just the
+# amount actually collected.
 HOME_INST = [
-    dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo"),
-    dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2"),
+    dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo",
+         pneta="p_neta", recarg="recargo", dpol="d_pol", com="com"),
+    dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2",
+         pneta="p_neta_2", recarg="recargo_2", dpol="d_pol_2", com="com_2"),
     dict(seq=3, amt="pago_subsec", cu="moned_3", d="fecha_pago_3", ck="no_cheque_3", cash="efectivo_3"),
     dict(seq=4, amt="pago_subsec", cu="moned_4", d="fecha_pago_4", ck="no_cheque_4", cash="efectivo_4"),
 ]
 HOME_FIELDS = dict(polno="no_poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
-                   forma="forma_pago", curcol="moned", pneta="p_neta", dpol="d_pol", com="com",
+                   forma="forma_pago", curcol="moned", pneta="p_neta", recarg="recargo",
+                   dpol="d_pol", com="com",
                    liquidada="liquidada", numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion")
-AUTO_SINGLE_INST = [dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")]
+AUTO_SINGLE_INST = [dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo",
+                         pneta="prima_neta", dpol="d_poliza")]
 
 CONFIGS = {
     "incendio":  dict(ptype="INCENDIO", idcol="num_id", fields={**HOME_FIELDS, "curcol": "moneda"},
@@ -157,7 +203,8 @@ CONFIGS = {
                                   forma="forma_pago", curcol="moneda", pneta="prima_neta", dpol="d_poliza",
                                   total="total", liquidada="liquidada", numliq="num_liquidacion",
                                   fliq="f_liquida1", renov="renovacion"),
-                      inst=[dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")],
+                      inst=[dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque",
+                                 cash="efectivo", pneta="prima_neta", dpol="d_poliza")],
                       veh=None, drv="licencias"),
 }
 
@@ -200,6 +247,10 @@ def main():
         consumed = {cfg["idcol"], *F.values()}
         for slot in cfg["inst"]:
             consumed |= {slot["amt"], slot["cu"], slot["d"], slot["ck"], slot["cash"]}
+            # The per-payment premium columns are now modeled, so they must
+            # leave the coveragesJson sweep — otherwise every recargo would be
+            # written twice, once as a column and once as a fake coverage.
+            consumed |= {slot[k] for k in ("pneta", "recarg", "dpol", "com") if slot.get(k)}
 
         for _, row in df.iterrows():
             cid = cust.get(norm_id(row[cfg["idcol"]]))
@@ -237,9 +288,11 @@ def main():
                 dt(row.get(F.get("desde", ""))) if F.get("desde") else None,
                 dt(row.get(F.get("hasta", ""))) if F.get("hasta") else None,
                 dec(row.get(F.get("pneta", ""))) if F.get("pneta") else None,
+                dec(row.get(F.get("recarg", ""))) if F.get("recarg") else None,
                 dec(row.get(F.get("dpol", ""))) if F.get("dpol") else None,
                 dec(row.get(F.get("com", ""))) if F.get("com") else None,
                 dec(row.get(F.get("total", ""))) if F.get("total") else None,
+                freq(row.get(F.get("forma", ""))) if F.get("forma") else None,
                 cur(row.get(F.get("curcol", ""))) if F.get("curcol") else "MXN",
                 s(row.get("observaciones")),
                 json.dumps(cov, ensure_ascii=False) if cov else None,
@@ -255,9 +308,12 @@ def main():
                 pdate = dt(row.get(slot["d"]))
                 if amt is None and pdate is None:
                     continue
+                # Slot breakdown, where the source table has one.
                 insts.append((str(uuid.uuid4()), pid, slot["seq"], amt,
                               cur(row.get(slot["cu"])), pdate, s(row.get(slot["ck"])),
-                              1 if truthy(row.get(slot["cash"])) else 0))
+                              1 if truthy(row.get(slot["cash"])) else 0,
+                              slot_dec(row, slot, "pneta"), slot_dec(row, slot, "recarg"),
+                              slot_dec(row, slot, "dpol"), slot_dec(row, slot, "com")))
 
             # vehicles
             def add_vehicle(make, model, body, engine, plate, year=None, state=None):
@@ -328,16 +384,19 @@ def main():
                        1 if truthy(r["concluido"]) else 0, s(r["resolucion"])))
 
     pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate,"
-                "policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations,"
+                "policyFrom,policyTo,netPremium,surcharge,policyFee,commission,total,paymentFrequency,"
+                "currency,observations,"
                 "coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,"
                 "legacySourceTable,legacyId,updatedAt")
-    ph = ",".join(["%s"] * 23)
+    ph = ",".join(["%s"] * 25)
     pol_upsert = (
         f"INSERT INTO policies ({pol_cols}) VALUES ({ph}) ON DUPLICATE KEY UPDATE "
         "customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),"
         "insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),"
-        "policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),"
-        "commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),"
+        "policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),"
+        "surcharge=VALUES(surcharge),policyFee=VALUES(policyFee),"
+        "commission=VALUES(commission),total=VALUES(total),paymentFrequency=VALUES(paymentFrequency),"
+        "currency=VALUES(currency),observations=VALUES(observations),"
         "coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),"
         "liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL")
 
@@ -399,8 +458,9 @@ def main():
 
 
     c.executemany("INSERT INTO policy_payment_installments "
-                  "(id,policyId,sequence,amount,currency,paidDate,checkNumber,isCash) "
-                  "VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", insts)
+                  "(id,policyId,sequence,amount,currency,paidDate,checkNumber,isCash,"
+                  "netPremium,surcharge,policyFee,commission) "
+                  "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", insts)
     c.executemany("INSERT INTO vehicles (id,customerId,policyId,make,model,modelYear,bodyType,"
                   "engineNumber,licensePlate,stateCode,legacySourceTable,legacyId) "
                   "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", vehicles)