Spaces:
Paused
Paused
| """ | |
| RewardPilot - Deterministic Scoring Engine | |
| ========================================== | |
| The engine sets the number; the LLM only narrates it. | |
| Given: | |
| - a set of cards the user holds (and optionally the full catalogue for discovery) | |
| - a transaction context (category, amount, merchant, brand offers) | |
| - month-to-date spend per card per category (to respect caps) | |
| - a light user persona | |
| ...it returns a point-value-adjusted ranking with transparent reasoning and | |
| the rupee value each card returns on this specific transaction. | |
| Design principles: | |
| 1. Everything is expressed in *net INR value*, never raw points. | |
| 2. Monthly caps are respected: accelerated rate applies only up to remaining cap. | |
| 3. Instant brand/issuer offers are added on top of rewards. | |
| 4. Annual fee is amortized as a tie-breaker / context signal, not a hard penalty | |
| on a single transaction. | |
| """ | |
| from dataclasses import dataclass, replace | |
| from typing import Dict, List, Optional | |
| from card_catalogue import Card, CATALOGUE_BY_ID, all_cards, canonical_card_id | |
| from rails import fee_for, fuel_waiver_inr, surcharge_known, FEE_GST_MULT | |
| import math | |
| def _r2(x: float) -> float: | |
| """Round money to 2 dp, half-up, in lockstep with engine.ts.""" | |
| # Decimal half-paisas such as 0.225 are commonly represented infinitesimally | |
| # below their mathematical value by binary floats. A tiny scaled epsilon | |
| # preserves the documented half-up rule instead of rounding them down. | |
| return math.floor(x * 100 + 0.5 + 1e-9) / 100 | |
| def apply_url_for(card_id: str) -> str: | |
| # Official issuer page when we have it; web search only as a last resort. | |
| from apply_links import APPLY_URL | |
| card = CATALOGUE_BY_ID.get(card_id) | |
| if card is not None and getattr(card, "closed", False): | |
| return "" | |
| direct = APPLY_URL.get(card_id) | |
| if direct: | |
| return direct | |
| import urllib.parse | |
| # Debit cards come with a bank account, so their fallback drops "credit card". | |
| kind_words = "apply" if card is not None and getattr(card, "kind", "credit") == "debit" else "credit card apply" | |
| q = urllib.parse.quote(f"{card.name if card else ''} {kind_words} {card.issuer if card else ''}") | |
| return f"https://www.google.com/search?q={q}" | |
| # Backward-compatible export for fuzz oracles. Product-specific terms live on | |
| # Card.upi_min_reward_amount; there is no universal transaction-size floor. | |
| UPI_MIN_REWARD_AMOUNT = 0 | |
| UPI_EXCLUDED_CATEGORIES = {"fuel", "rent", "wallet_load"} | |
| CONSERVATIVE_POINT_VALUE_INR = { | |
| "hdfc_infinia": 0.5, "hdfc_diners_black": 0.5, "hdfc_regalia_gold": 0.5, | |
| "icici_emeralde": 0.4, "axis_atlas": 0.5, "axis_magnus": 0.3, | |
| "amex_mrcc": 0.25, "amex_platinum_travel": 0.3, | |
| } | |
| def cash_point_value(card: Card) -> float: | |
| return 1.0 if card.reward_unit == "cashback" else CONSERVATIVE_POINT_VALUE_INR.get(card.id, card.point_value_inr) | |
| def _units_for_amount(card: Card, amount: float, rate: float, rail: str = "card", portal: bool = False) -> float: | |
| """Floor each transaction to the issuer's published earn block.""" | |
| block = ((getattr(card, "portal_earn_block_inr", None) or getattr(card, "earn_block_inr", None)) if portal else | |
| (getattr(card, "upi_earn_block_inr", None) or getattr(card, "earn_block_inr", None)) if rail == "upi" else | |
| getattr(card, "earn_block_inr", None)) | |
| eligible = math.floor(max(0.0, amount) / block) * block if block and block > 0 else max(0.0, amount) | |
| return eligible / 100.0 * rate | |
| def _portal_token(value) -> str: | |
| return "".join(ch for ch in str(value or "").lower() if ch.isalnum()) | |
| def _explicit_portal_attribution(card: Card, brand_key) -> bool: | |
| if not card.portal_name or not brand_key: | |
| return False | |
| brand = _portal_token(brand_key) | |
| portal = _portal_token(card.portal_name) | |
| aliases = (["smartbuy", "hdfcsmartbuy"] if "smartbuy" in portal else | |
| ["traveledge", "axistraveledge"] if "traveledge" in portal else | |
| ["ishop", "iciciishop"] if "ishop" in portal else | |
| ["hsbcrewards", "hsbcrewardsmarketplace"] if "rewardsmarketplace" in portal else | |
| ["rewardmultiplier", "amexrewardmultiplier"] if "rewardmultiplier" in portal else | |
| ["scapia"] if "scapia" in portal else [portal]) | |
| return brand in aliases | |
| class TxnContext: | |
| category: str | |
| amount: float | |
| merchant: Optional[str] = None | |
| brand_key: Optional[str] = None | |
| offers: Optional[List[Dict]] = None | |
| rail: str = "card" # "card" (POS/online) or "upi" (RuPay credit-on-UPI) | |
| channel: Optional[str] = None # "online" | "offline"; when known, filters offers by channel | |
| # TRIP SCOPE for travel scorings. None means "unknown, do not filter on it". | |
| # Enforced in score_transaction via offers.scope_filter, so the card | |
| # recommendation can no longer price a slab the channel rows refuse. | |
| international: Optional[bool] = None | |
| passengers: Optional[int] = None | |
| round_trip: Optional[bool] = None | |
| # card product id -> actual network printed on the held instance. Required | |
| # for products issued on multiple networks; absence fails closed. | |
| card_networks: Optional[Dict[str, str]] = None | |
| class CardScore: | |
| card_id: str | |
| card_name: str | |
| issuer: str | |
| network: str | |
| reward_cap_period: str | |
| reward_value_inr: float # rewards earned on this txn (capped) | |
| instant_offer_inr: float # instant discount from live offers | |
| deferred_cashback_inr: float # posts after payment; never reduces payable | |
| total_value_inr: float # reward + instant offer - issuer fee (rounded for display) | |
| raw_total: float # unrounded total - used for ranking | |
| effective_rate_pct: float # total value as % of spend | |
| capped: bool # whether monthly cap limited the reward | |
| reasons: List[str] | |
| held: bool # does the user hold this card | |
| point_value_inr_used: float = 1.0 | |
| point_value_inr_ceiling: float = 1.0 | |
| fee_known: bool = True | |
| # Best EMI-conditioned deal for this card, kept OUT of every total: the | |
| # discount only exists if the payment becomes a loan at checkout, so it is | |
| # reported beside the answer, never inside it. Defaults keep every | |
| # existing constructor call valid. | |
| emi_offer_inr: float = 0.0 | |
| emi_offer_text: Optional[str] = None | |
| emi_offer_code: Optional[str] = None | |
| # What this card charges to MAKE this payment, GST included: 1% on rent, | |
| # wallet loads, big utility bills and third-party education fees at most | |
| # issuers, and the fuel surcharge outside the waiver band. Netted out of | |
| # total_value_inr so a charging card cannot outrank a free one. | |
| fee_inr: float = 0.0 | |
| # The promo code the PRICED instant discount needs, when it needs one. | |
| # Mirror of CardScore.instantOfferCode in engine.ts. | |
| instant_offer_code: Optional[str] = None | |
| def _offer_applies(off: Dict, card: Card, allow_emi: bool = False, | |
| allow_unverified: bool = False) -> bool: | |
| # A term we never read - an uncapped % or a flat travel discount with no | |
| # minimum - is a discount we cannot compute, so unverified rows are refused | |
| # everywhere money is decided. channels.py's unverified-coupon line is the | |
| # ONE display path allowed through (allow_unverified=True) - it borrows this | |
| # function's card-matching so "your Axis card has a coupon here" is said only | |
| # to someone actually holding an Axis card, and offer_value()'s own zero for | |
| # these rows means even that path cannot price one by accident. Mirrored in | |
| # engine.ts offerApplies. | |
| from offers import is_unverified | |
| if is_unverified(off) and not allow_unverified: | |
| return False | |
| # The discount exists only if the payment is converted to EMI - a loan | |
| # decision we must never price in silently. Excluded from instant-offer | |
| # money; surfaced instead as a labelled EMI line (_emi_offer_value passes | |
| # allow_emi=True and requires the flag). Mirrored in engine.ts. | |
| if bool(off.get("requires_emi")) != allow_emi: | |
| return False | |
| # Claimable once, by someone who has never booked here. We cannot know that | |
| # about the user, so it is never priced as card value. Shown as a condition | |
| # instead (channels.py -> conditional_discount_inr). | |
| if off.get("requires_new_user"): | |
| return False | |
| # Confined to a payments app (BHIM etc): not claimable on a card at all. | |
| if off.get("requires_upi_app"): | |
| return False | |
| # Needs a code we could not name, so the user cannot apply it. | |
| if off.get("requires_promo_code") and not (off.get("promo_code") or "").strip(): | |
| return False | |
| # AN EXPLICIT CARD IS NOT A LICENCE TO SKIP THE REST. This used to RETURN on | |
| # the card match, jumping over the excluded-card, card-kind and network | |
| # gates below - so a row naming a card while also carrying requires_debit, | |
| # a network it is not on, AND that same card in its own excluded list | |
| # priced anyway. requires_debit is derived from evidence text after the | |
| # loader's contradiction check, so the feed can produce exactly that row. | |
| cid = off.get("card_id") | |
| if cid and canonical_card_id(cid) != card.id: | |
| return False | |
| iss = off.get("applies_to_issuer") or "" | |
| # An offer naming NO issuer and no card is open to anyone paying any way. | |
| # A blank issuer used to mean "any issuer", which handed every promo code | |
| # and platform sale to the whole catalogue as if plastic had earned it. | |
| # These are surfaced separately (offers.best_open_offer) so the user still | |
| # sees the money without it deciding which card to pull out. | |
| if not cid: | |
| if not iss: | |
| return False | |
| if iss != card.issuer: | |
| return False | |
| # "all HDFC Bank Credit Cards except Infinia & Diners Club Black" - the | |
| # offer is the issuer's, minus a named few. Mirrored in engine.ts. | |
| if card.id in [canonical_card_id(x) for x in (off.get("excluded_card_ids") or [])]: | |
| return False | |
| # Issuer-level offers still carry hidden card scoping the issuer field ignores: | |
| # a network limit ("Mastercard only") or a debit-card requirement. Enforce both, | |
| # else an HDFC-Mastercard-debit offer wrongly credits an HDFC-RuPay-credit card. | |
| # CARD KIND. The old comment claimed the catalogue is credit cards; that has | |
| # been false since the debit cards were added - there are 19. With no | |
| # inverse gate, an issuer-wide offer whose own evidence reads "ICICI Bank | |
| # CREDIT Card 25% discount" paid out on that issuer's DEBIT card, and could | |
| # make the debit card win the comparison outright. Both directions now hold. | |
| card_kind = getattr(card, "kind", "credit") | |
| kind_scope = off.get("card_kind_scope") or ("debit" if off.get("requires_debit") else "credit") | |
| if kind_scope != "both" and kind_scope != card_kind: | |
| return False | |
| net = (off.get("network") or "any").lower() | |
| if net != "any" and net != (card.network or "").lower(): | |
| return False | |
| return True | |
| def _instant_offer_value(card: Card, ctx: TxnContext): | |
| """Best applicable instant discount for this card on this txn (structured offers).""" | |
| from offers import offer_value | |
| best = 0.0 | |
| note = None | |
| # THE CODE COMES OUT TOO. _offer_applies prices a card-linked offer as soon | |
| # as its promo_code is non-empty, but this returned only the money - so 50 | |
| # active offers were priced into the headline with the code the user has to | |
| # type never surfaced anywhere. Mirror of engine.ts instantOffer. | |
| code = None | |
| cashback = False | |
| for off in (ctx.offers or []): | |
| if not _offer_applies(off, card): | |
| continue | |
| # channel filter: only when txn channel is known; 'both' always applies | |
| ch = getattr(ctx, "channel", None) | |
| off_ch = off.get("channel", "both") | |
| if ch and off_ch != "both" and off_ch != ch: | |
| continue | |
| if "type" in off: | |
| val = offer_value(off, ctx.amount, ctx.category) | |
| txt = off.get("text") or "" | |
| else: | |
| # legacy free-text offer (back-compat): parse "5%" / "₹5,000" | |
| import re | |
| txt = off.get("text", "") | |
| pct = re.search(r"(\d+(?:\.\d+)?)\s*%", txt) | |
| flat = re.search(r"(?:INR|Rs|₹)\s*([\d,]+)", txt) | |
| val = ctx.amount * float(pct.group(1)) / 100.0 if pct else (float(flat.group(1).replace(",", "")) if flat else 0.0) | |
| if val > best: | |
| best, note = val, txt | |
| code = (off.get("promo_code") or "").strip() or None | |
| cashback = off.get("type") == "cashback" | |
| return best, note, code, cashback | |
| def _emi_offer_value(card: Card, ctx: TxnContext): | |
| """Best EMI-conditioned discount for this card on this txn, with its code. | |
| Same claimability rules as instant offers (issuer/card/network/channel/ | |
| min-spend, and a code-requiring offer must have its code named), but ONLY | |
| offers flagged requires_emi. The value is NEVER added to card totals or | |
| net cost: it is returned separately so the UI can say "if you convert to | |
| EMI" in so many words. Mirrored in engine.ts.""" | |
| from offers import offer_value | |
| best, note, code = 0.0, None, None | |
| for off in (ctx.offers or []): | |
| if not _offer_applies(off, card, allow_emi=True): | |
| continue | |
| ch = getattr(ctx, "channel", None) | |
| off_ch = off.get("channel", "both") | |
| if ch and off_ch != "both" and off_ch != ch: | |
| continue | |
| val = offer_value(off, ctx.amount, ctx.category) if "type" in off else 0.0 | |
| if val > best: | |
| best = val | |
| note = off.get("text") or "" | |
| code = (off.get("promo_code") or "").strip() or None | |
| return best, note, code | |
| def _closed_or_debit(card_id: str) -> bool: | |
| """Cards that must never be offered as a DISCOVERY ("you should apply for | |
| this"). Mirror of engine.ts, which excludes both.""" | |
| c = CATALOGUE_BY_ID.get(card_id) | |
| if c is None: | |
| return False | |
| return bool(getattr(c, "closed", False)) or getattr(c, "kind", "credit") == "debit" | |
| def score_transaction( | |
| held_card_ids: List[str], | |
| ctx: TxnContext, | |
| mtd_spend: Optional[Dict[str, Dict[str, float]]] = None, | |
| include_discovery: bool = False, | |
| persona: Optional[List[str]] = None, | |
| ) -> Dict: | |
| """ | |
| Returns { | |
| held_ranked: [CardScore...], # user's own cards, best-first | |
| discovery: CardScore | None, # best card the user does NOT hold, if materially better | |
| context: {...} | |
| } | |
| mtd_spend: {card_id: {category: rupees_spent_this_month}} | |
| """ | |
| mtd_spend = mtd_spend or {} | |
| held_card_ids = list(dict.fromkeys(canonical_card_id(x) for x in held_card_ids)) | |
| # Legacy product state is the same physical card/cycle, not a second cap. | |
| for _old_id, _state in list(mtd_spend.items()): | |
| mtd_spend.setdefault(canonical_card_id(_old_id), _state) | |
| if ctx.card_networks: | |
| ctx = replace(ctx, card_networks={canonical_card_id(k): v for k, v in ctx.card_networks.items()}) | |
| # A BAD AMOUNT MUST NOT REACH THE REWARD MATH. engine.ts has clamped this | |
| # since the beginning; the server did not, so the same request that returned | |
| # Rs.0 on the device returned a NEGATIVE reward and a positive effective | |
| # rate from the API for amount = -10,000. | |
| try: | |
| _amt = float(ctx.amount) | |
| except (TypeError, ValueError): | |
| _amt = 0.0 | |
| if not (_amt > 0) or _amt != _amt or _amt == float("inf"): | |
| _amt = 0.0 | |
| if _amt != ctx.amount: | |
| ctx = replace(ctx, amount=_amt) | |
| # Trip scope applied ONCE, here, so every downstream consumer of ctx.offers | |
| # inherits it. compare_channels also pre-filters; the predicate is | |
| # idempotent, so neither can be the only guard. | |
| if ctx.offers: | |
| from offers import scope_filter | |
| ctx = replace(ctx, offers=scope_filter(ctx.offers, { | |
| "international": ctx.international, | |
| "passengers": ctx.passengers, | |
| "round_trip": ctx.round_trip, | |
| })) | |
| persona = persona or [] | |
| held_scores: List[CardScore] = [] | |
| all_scores: List[CardScore] = [] | |
| rail = getattr(ctx, "rail", "card") or "card" | |
| for catalogue_card in all_cards(): | |
| card = catalogue_card | |
| variants = list(getattr(card, "network_variants", None) or [card.network]) | |
| held = card.id in held_card_ids | |
| if held: | |
| requested = (ctx.card_networks or {}).get(card.id, "") | |
| selected = next((v for v in variants if v.lower() == str(requested).strip().lower()), None) | |
| card = replace( | |
| card, | |
| network=selected or "Unknown", | |
| upi_eligible=bool(selected and selected.lower() == "rupay"), | |
| ) | |
| # --- RuPay credit-on-UPI eligibility --- | |
| upi_block = None | |
| if rail == "upi": | |
| if not card.upi_eligible: | |
| upi_block = f"{card.network} cards can't be linked to UPI. Only RuPay credit cards work on UPI." | |
| elif ctx.amount < getattr(card, "upi_min_reward_amount", 0.0): | |
| upi_block = f"This card's UPI rewards start at ₹{card.upi_min_reward_amount:,.0f} per purchase." | |
| elif ctx.category in UPI_EXCLUDED_CATEGORIES: | |
| upi_block = f"{ctx.category.replace('_', ' ').title()} earns nothing on the UPI rail (excluded category)." | |
| # --- reward value, respecting monthly caps --- | |
| txn_channel = getattr(ctx, "channel", None) | |
| # Missing import/payment provenance is not proof that the purchase was | |
| # online. Cards with online-only accelerators therefore fall back to | |
| # their guaranteed offline/base tier until the channel is explicit. | |
| if txn_channel not in ("online", "offline") and card.online_boost_only: | |
| txn_channel = "offline" | |
| if upi_block: | |
| rate = 0.0 | |
| elif rail == "upi": | |
| rate = card.upi_effective_rate(ctx.category, ctx.brand_key, ctx.amount, txn_channel) | |
| else: | |
| rate = card.effective_rate(ctx.category, ctx.brand_key, ctx.amount, txn_channel) | |
| gross_units = _units_for_amount(card, ctx.amount, rate, rail) | |
| base_units_earned = 0.0 | |
| capped = False | |
| # Which monthly cap applies, in priority: 1) a BRAND cap (when the rate came | |
| # from a brand bonus, e.g. Millennia 5% on Amazon/Swiggy... INR 1,000/mo shared | |
| # across brands), 2) a category CAP GROUP (e.g. HSBC INR 1,000/mo), 3) per-category. | |
| using_brand = (not card.boost_suppressed(txn_channel) | |
| and ctx.brand_key is not None and ctx.brand_key in card.brand_bonuses | |
| and ctx.amount >= card.brand_min_txn.get(ctx.brand_key, 0) | |
| and card.brand_bonuses[ctx.brand_key] >= card.category_rates.get(ctx.category, card.base_rate)) | |
| brand_grp = next((g for g in card.brand_caps if ctx.brand_key in g.get("brands", [])), None) if using_brand else None | |
| cat_grp = (None if card.boost_suppressed(txn_channel) and card.online_boost_only else | |
| next((g for g in card.cap_groups if ctx.category in g.get("categories", [])), None)) | |
| cap_bucket_key = None | |
| brand_txn_cap = card.brand_txn_caps.get(ctx.brand_key) if using_brand and ctx.brand_key else None | |
| if brand_txn_cap is not None: | |
| gross_units = min(gross_units, brand_txn_cap) | |
| if rate > card.base_rate and not upi_block and card.accelerated_cap_units is not None: | |
| cap_units = card.accelerated_cap_units | |
| cap_bucket_key = "__acceleratedUnits" | |
| elif brand_grp: | |
| cap_units = brand_grp["cap"] | |
| cap_bucket_key = "__cap:brand:" + "|".join(brand_grp["brands"]) | |
| elif cat_grp: | |
| cap_units = cat_grp["cap"] | |
| cap_bucket_key = "__cap:cat:" + "|".join(cat_grp["categories"]) | |
| else: | |
| cap_units = card.caps.get(ctx.category) | |
| if cap_units is not None: | |
| cap_bucket_key = "__cap:single:" + ctx.category | |
| if not upi_block and cap_units is not None and rate > 0: | |
| # WHAT THE CAP HAS ALREADY EATEN, valued at each member's OWN rate. | |
| # Prior spend is stored in rupees and used to be converted back at | |
| # THIS transaction's rate, which is only right when every member of | |
| # the bucket earns the same. Axis ACE shares one Rs.500/month cap | |
| # across bill payments at 5% and food delivery at 4%: after the cap | |
| # was fully spent on bills, a Swiggy order still paid out, because | |
| # the same rupees re-read at 4% looked like less of the cap. | |
| if mtd_spend.get(card.id, {}).get("__bucketState") and cap_bucket_key: | |
| already_units = mtd_spend.get(card.id, {}).get(cap_bucket_key, 0.0) | |
| elif brand_grp: | |
| already_units = sum( | |
| mtd_spend.get(card.id, {}).get("~" + b, 0.0) / 100.0 | |
| * (card.brand_bonuses or {}).get(b, card.base_rate) | |
| for b in brand_grp["brands"]) | |
| elif cat_grp: | |
| already_units = sum( | |
| mtd_spend.get(card.id, {}).get(c, 0.0) / 100.0 | |
| * (0.0 if c in card.excluded_categories else card.category_rates.get(c, card.base_rate)) | |
| for c in cat_grp["categories"]) | |
| else: | |
| already_units = mtd_spend.get(card.id, {}).get(ctx.category, 0.0) / 100.0 * rate | |
| remaining_units = max(0.0, cap_units - already_units) | |
| if gross_units > remaining_units: | |
| accel_units = remaining_units | |
| # WHAT SPEND BEYOND THE CAP EARNS. Falling back to the base rate | |
| # made the cap ARITHMETICALLY INERT wherever the capped category | |
| # had no accelerated rate of its own - 28 cards are in that | |
| # position. HDFC caps Infinia grocery and utility earn at 2,000 | |
| # points a month; since utilities earn the base rate, "cap the | |
| # first 2,000 then pay base on the rest" reconstitutes the | |
| # uncapped total exactly, and Rs.2,00,000 of utility spend | |
| # returned Rs.6,600 against a Rs.2,000 ceiling. A cap on the | |
| # BASE rate is a hard cap; a cap on an ACCELERATED rate still | |
| # drops to base, which is the real rule on Millennia. | |
| base_units_value = 0.0 | |
| if ctx.category not in card.excluded_categories and rate > card.base_rate: | |
| base_units_value = card.base_rate | |
| rupees_at_accel = remaining_units / rate * 100.0 if rate > 0 else 0.0 | |
| rupees_beyond = max(0.0, ctx.amount - rupees_at_accel) | |
| base_units_earned = _units_for_amount(card, rupees_beyond, base_units_value, rail) | |
| gross_units = accel_units + base_units_earned | |
| capped = True | |
| elif rate <= card.base_rate: | |
| base_units_earned = gross_units | |
| elif not upi_block and rate <= card.base_rate: | |
| base_units_earned = gross_units | |
| # Products such as SBI Cashback cap the accelerated 5% tier and the | |
| # ordinary/fallback 1% tier independently. Clamp only the latter here; | |
| # combining both ceilings or leaving the fallback uncapped overstates a | |
| # Rs.10 lakh online purchase by thousands. | |
| if not upi_block and card.base_cap_units is not None and base_units_earned > 0: | |
| remain_base = max(0.0, card.base_cap_units - mtd_spend.get(card.id, {}).get("__baseUnits", 0.0)) | |
| if base_units_earned > remain_base: | |
| gross_units -= base_units_earned - remain_base | |
| base_units_earned = remain_base | |
| capped = True | |
| # UPI-rail monthly reward-unit cap (e.g. Tata Neu 500 NeuCoins/mo on UPI). Hard cap: | |
| # units beyond it earn nothing. Accumulated UPI units live under the "__upiUnits" key. | |
| if not upi_block and rail == "upi" and card.upi_reward_cap_units is not None and gross_units > 0: | |
| remain_upi = max(0.0, card.upi_reward_cap_units - mtd_spend.get(card.id, {}).get("__upiUnits", 0.0)) | |
| if gross_units > remain_upi: | |
| gross_units = remain_upi | |
| capped = True | |
| if not upi_block and card.global_cap_units is not None and gross_units > 0: | |
| remain_global = max(0.0, card.global_cap_units - mtd_spend.get(card.id, {}).get("__globalUnits", 0.0)) | |
| if gross_units > remain_global: | |
| ratio = remain_global / gross_units if gross_units else 0.0 | |
| gross_units = remain_global; base_units_earned *= ratio; capped = True | |
| annual_grp = next((g for g in card.brand_annual_caps if ctx.brand_key in g.get("brands", [])), None) if using_brand else None | |
| if not upi_block and annual_grp and gross_units > 0: | |
| key = "__annual:brand:" + "|".join(annual_grp["brands"]) | |
| remain_annual = max(0.0, annual_grp["cap"] - mtd_spend.get(card.id, {}).get(key, 0.0)) | |
| if gross_units > remain_annual: | |
| ratio = remain_annual / gross_units if gross_units else 0.0 | |
| gross_units = remain_annual; base_units_earned *= ratio; capped = True | |
| point_value_inr_used = cash_point_value(card) | |
| reward_inr = gross_units * point_value_inr_used | |
| # --- instant offers (don't apply on the UPI rail) --- | |
| if rail == "upi": | |
| offer_value_inr, offer_note, offer_code, offer_is_cashback = 0.0, None, None, False | |
| emi_inr, emi_note, emi_code = 0.0, None, None | |
| else: | |
| offer_value_inr, offer_note, offer_code, offer_is_cashback = _instant_offer_value(card, ctx) | |
| emi_inr, emi_note, emi_code = _emi_offer_value(card, ctx) | |
| offer_inr = 0.0 if offer_is_cashback else offer_value_inr | |
| deferred_cashback_inr = offer_value_inr if offer_is_cashback else 0.0 | |
| # WHAT THE CARD CHARGES TO MAKE THIS PAYMENT, GST included. Netted out | |
| # of the total so an issuer that charges 1% on rent can no longer | |
| # outrank one that charges nothing. Thresholds are cumulative per | |
| # STATEMENT CYCLE, so same-category spend already through this card in | |
| # this cycle decides how much of THIS payment sits above the line. | |
| # Reward caps and issuer fee thresholds do not necessarily share a | |
| # reset clock. Historical replay writes fee-cycle spend under a | |
| # dedicated key; keep the legacy category fallback for older clients. | |
| prior_cycle = mtd_spend.get(card.id, {}).get( | |
| "__fee:" + ctx.category, | |
| mtd_spend.get(card.id, {}).get(ctx.category, 0.0), | |
| ) | |
| direct_education = ctx.category == "education" and not ctx.brand_key | |
| fee_unknown = (rail == "card" and ctx.category != "fuel" and not direct_education | |
| and ctx.category in {"rent", "wallet_load", "utilities", "education"} | |
| and not surcharge_known(card.id, ctx.category)) | |
| fee_inr = (0.0 if rail == "upi" else | |
| 0.0 if ctx.category == "education" and not ctx.brand_key else | |
| _r2(ctx.amount * 0.01 * FEE_GST_MULT) if fee_unknown else | |
| fee_for(card.id, ctx.category, ctx.amount, prior_cycle, ctx.brand_key, | |
| mtd_spend.get(card.id, {}).get("__fuelWaiverInr", 0.0))) | |
| # _r2, not round(): Python's round is banker's and JS Math.round is | |
| # half-up, which put the two engines a paisa apart on 448 scorings. | |
| total = _r2(reward_inr + offer_inr + deferred_cashback_inr - fee_inr) | |
| eff_pct = (total / ctx.amount * 100.0) if ctx.amount else 0.0 | |
| # --- reasons --- | |
| reasons: List[str] = [] | |
| if upi_block: | |
| reasons.append(upi_block) | |
| elif ctx.category in card.excluded_categories: | |
| reasons.append(f"{ctx.category.replace('_', ' ').title()} is excluded on this card, so it earns nothing.") | |
| elif card.boost_suppressed(txn_channel): | |
| # Honest in-store framing: say WHY the number is the base rate. | |
| reasons.append( | |
| f"Earns {rate:g} {card.reward_unit}/₹100 in-store = ₹{reward_inr:,.0f} back. " | |
| f"This card's higher rates apply to online spends only." | |
| ) | |
| else: | |
| reasons.append( | |
| f"Earns {rate:g} {card.reward_unit}/₹100 in {ctx.category.replace('_',' ')} " | |
| f"(₹{card.point_value_inr:g}/{card.reward_unit[:-1] if card.reward_unit.endswith('s') else card.reward_unit}) " | |
| f"= ₹{reward_inr:,.0f} back." | |
| ) | |
| if not upi_block and card.reward_unit != "cashback" and reward_inr > 0: | |
| ceiling = (f"; selected premium redemptions can reach ₹{card.point_value_inr:g}/point" | |
| if card.point_value_inr > point_value_inr_used else "") | |
| reasons.append(f"Ranked at a conservative ₹{point_value_inr_used:g}/point{ceiling}.") | |
| if offer_value_inr > 0 and offer_note: | |
| label = "After payment" if offer_is_cashback else "Live offer" | |
| reasons.append(f"{label}: {offer_note} (~₹{offer_value_inr:,.0f}).") | |
| if fee_unknown: | |
| reasons.insert(0, f"Issuer fee is not verified for this card. Ranking reserves an estimated 1% plus GST (₹{fee_inr:,.0f}) instead of treating the unknown as free.") | |
| # Portal-only elevated rate (HDFC SmartBuy, Axis Travel Edge) - informational, | |
| # NOT scored, since it isn't earned on a direct merchant booking. | |
| portal_rate = card.portal_rates.get(ctx.category) if not upi_block else None | |
| if portal_rate and portal_rate > rate and ctx.category not in card.excluded_categories: | |
| # Portal earn is unit-based and capped monthly (SmartBuy 15k RP/mo, iShop 18k/mo); | |
| # redeem at the portal point value. Show the TRUE effective %, never the raw unit rate. | |
| portal_pv = card.portal_point_value_inr if card.portal_point_value_inr is not None else card.point_value_inr | |
| gross_units = _units_for_amount(card, ctx.amount, portal_rate, "card", portal=True) | |
| portal_state = mtd_spend.get(card.id, {}) | |
| portal_usage_known = card.portal_cap_units is None or "__portalUnits" in portal_state | |
| prior_portal_units = float(portal_state.get("__portalUnits", 0.0)) if portal_usage_known else 0.0 | |
| remaining_portal = (gross_units if card.portal_cap_units is None else | |
| max(0.0, card.portal_cap_units - prior_portal_units)) | |
| portal_units = min(gross_units, remaining_portal) | |
| portal_inr = portal_units * portal_pv | |
| portal_capped = card.portal_cap_units is not None and gross_units > remaining_portal | |
| eff_portal_pct = (portal_inr / ctx.amount * 100.0) if ctx.amount else 0.0 | |
| if not portal_usage_known: | |
| up_to = min(gross_units, card.portal_cap_units) * portal_pv | |
| reasons.append( | |
| f"Booking via {card.portal_name or 'the issuer portal'} could return up to ₹{up_to:,.0f}, " | |
| "but this month's portal-cap usage is unverified, so it is not ranked above guaranteed value." | |
| ) | |
| elif portal_inr > total: | |
| cap_note = f", capped at the portal's monthly {card.portal_cap_units:,.0f}-point limit" if portal_capped else "" | |
| reasons.append( | |
| f"Booking via {card.portal_name or 'the issuer portal'} would earn " | |
| f"about {round(eff_portal_pct, 1):g}% here (₹{portal_inr:,.0f}){cap_note}." | |
| ) | |
| if capped: | |
| reasons.append("Monthly accelerated cap partly reached. Value above the cap drops to the base rate.") | |
| sc = CardScore( | |
| card_id=card.id, | |
| card_name=card.name, | |
| issuer=card.issuer, | |
| network=card.network, | |
| reward_cap_period=card.reward_cap_period or "unknown", | |
| reward_value_inr=_r2(reward_inr), | |
| instant_offer_inr=_r2(offer_inr), | |
| deferred_cashback_inr=_r2(deferred_cashback_inr), | |
| instant_offer_code=offer_code, | |
| total_value_inr=_r2(total), | |
| raw_total=total, | |
| effective_rate_pct=_r2(eff_pct), | |
| capped=capped, | |
| reasons=reasons, | |
| held=held, | |
| point_value_inr_used=point_value_inr_used, | |
| point_value_inr_ceiling=card.point_value_inr, | |
| emi_offer_inr=_r2(emi_inr), | |
| emi_offer_text=emi_note, | |
| emi_offer_code=emi_code, | |
| fee_inr=_r2(fee_inr), | |
| fee_known=not fee_unknown, | |
| ) | |
| all_scores.append(sc) | |
| if held: | |
| held_scores.append(sc) | |
| # sort best-first by the unrounded total (display rounding must not flip ties). | |
| # On the UPI rail a non-RuPay card can't be linked at all, so it must never rank above a | |
| # UPI-eligible card of equal value (e.g. a sub-₹2,000 UPI spend earns everyone ₹0 - the top | |
| # pick should still be a card you can actually pay with on UPI). | |
| def _rank_key(s): | |
| demerit = 1 if (rail == "upi" and str(s.network).lower() != "rupay") else 0 | |
| # Card id is the FINAL tie-break. Without it a tie was broken by | |
| # catalogue insertion order, and card_catalogue.py and cards.ts list the | |
| # same 139 ids in different order - so the server and the app named | |
| # DIFFERENT winners for the same tied purchase. Mirror of engine.ts. | |
| return (demerit, -s.raw_total, s.card_id) | |
| held_scores.sort(key=_rank_key) | |
| all_scores.sort(key=_rank_key) | |
| # --- personalized, comparative reasoning for why each card ranks where it does --- | |
| def _money(n: float) -> str: | |
| return "₹{:,.0f}".format(n) | |
| if held_scores: | |
| top = held_scores[0] | |
| none_earn = top.total_value_inr <= 0 | |
| for i, s in enumerate(held_scores): | |
| lead = [] | |
| if i == 0: | |
| if none_earn: | |
| lead.append( | |
| "None of your cards earn rewards on this UPI payment. Here's why." | |
| if rail == "upi" else | |
| "None of your cards earn rewards on this purchase." | |
| ) | |
| elif len(held_scores) > 1: | |
| d = top.total_value_inr - held_scores[1].total_value_inr | |
| if d > 0: | |
| lead.append( | |
| f"Your top card here, returning {_money(d)} " | |
| f"more than your next best ({held_scores[1].card_name})." | |
| ) | |
| else: | |
| lead.append(f"Tied with {held_scores[1].card_name}; both earn the same here.") | |
| else: | |
| lead.append("Your best card for this purchase.") | |
| elif not none_earn and s.total_value_inr > 0: | |
| d = top.total_value_inr - s.total_value_inr | |
| if d > 0: | |
| lead.append(f"Ranks #{i+1}, {_money(d)} less than {top.card_name} on this spend.") | |
| else: | |
| lead.append(f"Ties {top.card_name} at the same value.") | |
| s.reasons = lead + s.reasons | |
| # discovery: best non-held card that beats the user's best held card by a margin | |
| discovery = None | |
| if include_discovery and held_scores: | |
| best_held = held_scores[0].total_value_inr | |
| for sc in all_scores: | |
| # "Apply for this card" must never name a card no one can apply for, | |
| # nor a DEBIT card (you do not choose a debit card for its rewards). | |
| # engine.ts filtered both; this rail filtered neither, so the server | |
| # recommended Axis Atlas - closed to new applicants since Sep 2025. | |
| if _closed_or_debit(sc.card_id): | |
| continue | |
| if not sc.held and sc.total_value_inr > best_held * 1.10: # >10% better | |
| discovery = sc | |
| break | |
| elif include_discovery and not held_scores: | |
| discovery = next((s for s in all_scores if not s.held and not _closed_or_debit(s.card_id)), None) | |
| # --- richer reasoning for the better card the user does NOT hold --- | |
| if discovery: | |
| dcard = CATALOGUE_BY_ID.get(discovery.card_id) | |
| best_held_val = held_scores[0].total_value_inr if held_scores else 0.0 | |
| extra = discovery.total_value_inr - best_held_val | |
| r: List[str] = [] | |
| if held_scores: | |
| r.append( | |
| f"Beats your best card here by {_money(extra)} " | |
| f"({_money(discovery.total_value_inr)} vs {_money(best_held_val)} on {held_scores[0].card_name})." | |
| ) | |
| else: | |
| r.append(f"Would return {_money(discovery.total_value_inr)} on this purchase.") | |
| if dcard: | |
| d_rate = dcard.category_rates.get(ctx.category, dcard.base_rate) | |
| if ctx.category not in dcard.excluded_categories: | |
| r.append(f"Earns {d_rate:g} {dcard.reward_unit}/₹100 on {ctx.category.replace('_',' ')}.") | |
| if dcard.highlights: | |
| r.append(dcard.highlights[0] + ".") | |
| r.append(f"Annual fee: {dcard.annual_fee_text}.") | |
| discovery.reasons = r | |
| # savings vs best held: difference between top held and the worst reasonable choice | |
| savings_vs_worst = 0.0 | |
| if len(held_scores) >= 2: | |
| savings_vs_worst = _r2(held_scores[0].total_value_inr - held_scores[-1].total_value_inr) | |
| return { | |
| "held_ranked": [s.__dict__ for s in held_scores], | |
| "discovery": discovery.__dict__ if discovery else None, | |
| "savings_vs_worst": savings_vs_worst, | |
| "context": { | |
| "category": ctx.category, | |
| "amount": ctx.amount, | |
| "merchant": ctx.merchant, | |
| "brand_key": ctx.brand_key, | |
| "offers": ctx.offers or [], | |
| "rail": rail, | |
| }, | |
| } | |
| def accumulate_spend(mtd: Dict[str, Dict[str, float]], card_id: str, txn: Dict) -> None: | |
| """Replay the actual earned-unit buckets for one attributed transaction. | |
| Category/brand rupees remain for fee thresholds and legacy callers, while | |
| named buckets preserve the rail/channel/minimum-sensitive reward units that | |
| cannot be reconstructed from rupees later. | |
| """ | |
| card = CATALOGUE_BY_ID.get(card_id) | |
| if card is None: | |
| return | |
| try: | |
| amount = max(0.0, float(txn.get("amount") or 0.0)) | |
| except (TypeError, ValueError): | |
| amount = 0.0 | |
| category = str(txn.get("category") or "general") | |
| brand = txn.get("brand_key") | |
| rail = txn.get("rail") if txn.get("rail") in ("card", "upi") else "card" | |
| # Missing history provenance never authorises an online-only accelerator. | |
| channel = txn.get("channel") if txn.get("channel") in ("online", "offline") else ("offline" if card.online_boost_only else None) | |
| if rail == "upi": | |
| blocked = (not card.upi_eligible or amount < card.upi_min_reward_amount or category in UPI_EXCLUDED_CATEGORIES) | |
| rate = 0.0 if blocked else card.upi_effective_rate(category, brand, amount, channel) | |
| else: | |
| blocked = False | |
| rate = card.effective_rate(category, brand, amount, channel) | |
| gross = _units_for_amount(card, amount, rate, rail) | |
| using_brand = (not card.boost_suppressed(channel) and brand in card.brand_bonuses | |
| and amount >= card.brand_min_txn.get(brand, 0) | |
| and card.brand_bonuses[brand] >= card.category_rates.get(category, card.base_rate)) | |
| brand_grp = next((g for g in card.brand_caps if brand in g.get("brands", [])), None) if using_brand else None | |
| cat_grp = (None if card.boost_suppressed(channel) and card.online_boost_only else | |
| next((g for g in card.cap_groups if category in g.get("categories", [])), None)) | |
| cap = None | |
| bucket = None | |
| if using_brand and brand and card.brand_txn_caps.get(brand) is not None: | |
| gross = min(gross, card.brand_txn_caps[brand]) | |
| if rate > card.base_rate and not blocked and card.accelerated_cap_units is not None: | |
| cap = float(card.accelerated_cap_units); bucket = "__acceleratedUnits" | |
| elif brand_grp: | |
| cap = float(brand_grp["cap"]); bucket = "__cap:brand:" + "|".join(brand_grp["brands"]) | |
| elif cat_grp: | |
| cap = float(cat_grp["cap"]); bucket = "__cap:cat:" + "|".join(cat_grp["categories"]) | |
| elif card.caps.get(category) is not None: | |
| cap = float(card.caps[category]); bucket = "__cap:single:" + category | |
| state = mtd.setdefault(card_id, {}) | |
| fuel_waived = (fuel_waiver_inr(card_id, amount, brand, state.get("__fuelWaiverInr", 0.0)) | |
| if category == "fuel" and rail == "card" else 0.0) | |
| prior_cap = state.get(bucket, 0.0) if bucket else 0.0 | |
| remaining = max(0.0, cap - prior_cap) if cap is not None else None | |
| cap_consumed = min(gross, remaining) if remaining is not None else 0.0 | |
| base_units = gross if (not blocked and rate <= card.base_rate) else 0.0 | |
| if remaining is not None and gross > remaining: | |
| if rate > card.base_rate and category not in card.excluded_categories: | |
| beyond_rupees = max(0.0, amount - (remaining / rate * 100.0 if rate else 0.0)) | |
| base_units = _units_for_amount(card, beyond_rupees, card.base_rate, rail) | |
| gross = remaining + base_units | |
| else: | |
| gross = remaining | |
| base_units = remaining if rate <= card.base_rate else 0.0 | |
| if card.base_cap_units is not None and base_units > 0: | |
| remain_base = max(0.0, card.base_cap_units - state.get("__baseUnits", 0.0)) | |
| if base_units > remain_base: | |
| gross -= base_units - remain_base | |
| base_units = remain_base | |
| if rail == "upi" and card.upi_reward_cap_units is not None and gross > 0: | |
| remain_upi = max(0.0, card.upi_reward_cap_units - state.get("__upiUnits", 0.0)) | |
| if gross > remain_upi: | |
| ratio = remain_upi / gross if gross else 0.0 | |
| gross = remain_upi; base_units *= ratio; cap_consumed *= ratio | |
| if card.global_cap_units is not None and gross > 0: | |
| remain_global = max(0.0, card.global_cap_units - state.get("__globalUnits", 0.0)) | |
| if gross > remain_global: | |
| ratio = remain_global / gross if gross else 0.0 | |
| gross = remain_global; base_units *= ratio; cap_consumed *= ratio | |
| annual_grp = next((g for g in card.brand_annual_caps if brand in g.get("brands", [])), None) if using_brand else None | |
| if annual_grp and gross > 0: | |
| annual_key = "__annual:brand:" + "|".join(annual_grp["brands"]) | |
| remain_annual = max(0.0, annual_grp["cap"] - state.get(annual_key, 0.0)) | |
| if gross > remain_annual: | |
| ratio = remain_annual / gross if gross else 0.0 | |
| gross = remain_annual; base_units *= ratio; cap_consumed *= ratio | |
| state["__bucketState"] = 1.0 | |
| state[category] = state.get(category, 0.0) + amount | |
| if brand: | |
| state["~" + brand] = state.get("~" + brand, 0.0) + amount | |
| if bucket and cap_consumed > 0: | |
| state[bucket] = state.get(bucket, 0.0) + cap_consumed | |
| if base_units > 0: | |
| state["__baseUnits"] = state.get("__baseUnits", 0.0) + base_units | |
| if gross > 0: | |
| state["__globalUnits"] = state.get("__globalUnits", 0.0) + gross | |
| if annual_grp and gross > 0: | |
| state[annual_key] = state.get(annual_key, 0.0) + gross | |
| if rail == "upi" and gross > 0: | |
| state["__upiUnits"] = state.get("__upiUnits", 0.0) + gross | |
| if fuel_waived > 0: | |
| state["__fuelWaiverInr"] = state.get("__fuelWaiverInr", 0.0) + fuel_waived | |
| if rail == "card" and _explicit_portal_attribution(card, brand): | |
| portal_rate = (card.portal_rates or {}).get(category) | |
| if portal_rate: | |
| gross_portal = _units_for_amount(card, amount, portal_rate, "card", portal=True) | |
| remaining_portal = (gross_portal if card.portal_cap_units is None else | |
| max(0.0, card.portal_cap_units - state.get("__portalUnits", 0.0))) | |
| state["__portalUnits"] = state.get("__portalUnits", 0.0) + min(gross_portal, remaining_portal) | |
| def _history_month(txn: Dict) -> str: | |
| date = str(txn.get("date") or "") | |
| return date[:7] if len(date) >= 7 else "_" | |
| def _history_statement_period(txn: Dict, day) -> str: | |
| fallback = _history_month(txn) | |
| try: | |
| close_day = int(day) | |
| except (TypeError, ValueError): | |
| return fallback | |
| date = str(txn.get("date") or "") | |
| if close_day < 1 or close_day > 28 or len(date) < 10: | |
| return fallback | |
| try: | |
| year, month, txn_day = int(date[:4]), int(date[5:7]), int(date[8:10]) | |
| except (TypeError, ValueError): | |
| return fallback | |
| if not (1 <= month <= 12 and 1 <= txn_day <= 31): | |
| return fallback | |
| end_month = year * 12 + month - 1 + (1 if txn_day > close_day else 0) | |
| return f"s:{end_month // 12}-{end_month % 12 + 1:02d}" | |
| def _history_bucket(store: Dict, card_id: str, key: str) -> Dict[str, float]: | |
| return store.setdefault(card_id, {}).setdefault(key, {}) | |
| def _historical_mtd_for(card: Card, txn: Dict, state: Dict, statement_days: Dict) -> Dict: | |
| reward_key = (_history_statement_period(txn, statement_days.get(card.id)) | |
| if card.reward_cap_period == "statement_cycle" else _history_month(txn)) | |
| fee_key = _history_statement_period(txn, statement_days.get(card.id)) | |
| date = str(txn.get("date") or "") | |
| annual_key = date[:4] if len(date) >= 4 else "_" | |
| merged = dict(_history_bucket(state["reward"], card.id, reward_key)) | |
| for key, value in _history_bucket(state["fee"], card.id, fee_key).items(): | |
| if key == "__fuelWaiverInr": | |
| merged[key] = value | |
| elif not key.startswith("__") and not key.startswith("~"): | |
| merged["__fee:" + key] = value | |
| for key, value in _history_bucket(state["annual"], card.id, annual_key).items(): | |
| if key.startswith("__annual:"): | |
| merged[key] = value | |
| portal = _history_bucket(state["portal"], card.id, _history_month(txn)) | |
| if "__portalUnits" in portal: | |
| merged["__portalUnits"] = portal["__portalUnits"] | |
| return {card.id: merged} | |
| def _accumulate_historical_spend(card: Card, txn: Dict, state: Dict, statement_days: Dict) -> None: | |
| try: | |
| amount = max(0.0, float(txn.get("amount") or 0.0)) | |
| except (TypeError, ValueError): | |
| amount = 0.0 | |
| if amount <= 0: | |
| return | |
| reward_key = (_history_statement_period(txn, statement_days.get(card.id)) | |
| if card.reward_cap_period == "statement_cycle" else _history_month(txn)) | |
| fee_key = _history_statement_period(txn, statement_days.get(card.id)) | |
| date = str(txn.get("date") or "") | |
| annual_key = date[:4] if len(date) >= 4 else "_" | |
| reward_state = _history_bucket(state["reward"], card.id, reward_key) | |
| fee_state = _history_bucket(state["fee"], card.id, fee_key) | |
| annual_state = _history_bucket(state["annual"], card.id, annual_key) | |
| portal_state = _history_bucket(state["portal"], card.id, _history_month(txn)) | |
| # Reuse the single-transaction unit oracle against the exact merged ledger, | |
| # then distribute its typed outputs back to their independent clocks. | |
| visible = _historical_mtd_for(card, txn, state, statement_days) | |
| replay = {card.id: dict(visible[card.id])} | |
| accumulate_spend(replay, card.id, txn) | |
| after = replay[card.id] | |
| category = str(txn.get("category") or "general") | |
| brand = txn.get("brand_key") | |
| reward_keys = { | |
| "__bucketState", "__baseUnits", "__globalUnits", "__upiUnits", "__acceleratedUnits", | |
| } | |
| for key, value in after.items(): | |
| if key == category or (brand and key == "~" + str(brand)) or key in reward_keys or key.startswith("__cap:"): | |
| reward_state[key] = value | |
| fee_state[category] = fee_state.get(category, 0.0) + amount | |
| if brand: | |
| fee_state["~" + str(brand)] = fee_state.get("~" + str(brand), 0.0) + amount | |
| if "__fuelWaiverInr" in after: | |
| fee_state["__fuelWaiverInr"] = after["__fuelWaiverInr"] | |
| for key, value in after.items(): | |
| if key.startswith("__annual:"): | |
| annual_state[key] = value | |
| if "__portalUnits" in after: | |
| portal_state["__portalUnits"] = after["__portalUnits"] | |
| def analyse_transaction_history( | |
| held_card_ids: List[str], | |
| transactions: List[Dict], | |
| statement_days: Optional[Dict[str, int]] = None, | |
| held_networks: Optional[Dict[str, str]] = None, | |
| ) -> Dict: | |
| """ | |
| For each past transaction, determine whether the card used was optimal, | |
| and how much was left on the table. | |
| transactions: [{id, date, merchant, category, amount, card_id_used}] | |
| """ | |
| results = [None] * len(transactions) | |
| total_lost = 0.0 | |
| held_card_ids = list(dict.fromkeys(canonical_card_id(card_id) for card_id in held_card_ids)) | |
| statement_days = {canonical_card_id(key): value for key, value in (statement_days or {}).items()} | |
| held_networks = {canonical_card_id(key): value for key, value in (held_networks or {}).items()} | |
| history_state = {"reward": {}, "fee": {}, "annual": {}, "portal": {}} | |
| order = sorted(range(len(transactions)), key=lambda i: ( | |
| str(transactions[i].get("date") or ""), str(transactions[i].get("id") or ""), i | |
| )) | |
| for index in order: | |
| txn = transactions[index] | |
| mtd = {} | |
| for card_id in held_card_ids: | |
| card = CATALOGUE_BY_ID.get(card_id) | |
| if card is not None: | |
| mtd.update(_historical_mtd_for(card, txn, history_state, statement_days)) | |
| def _ctx(rail): | |
| return TxnContext( | |
| category=txn["category"], amount=float(txn["amount"]), | |
| merchant=txn.get("merchant"), brand_key=txn.get("brand_key"), | |
| offers=txn.get("offers"), rail=rail, channel=txn.get("channel"), | |
| card_networks=held_networks, | |
| ) | |
| # statement transactions were card swipes -> "used" is on the card rail; | |
| # also check the UPI (RuPay) rail to find the true cross-rail optimum. | |
| scored_card = score_transaction(held_card_ids, _ctx("card"), mtd_spend=mtd, include_discovery=False) | |
| scored_upi = score_transaction(held_card_ids, _ctx("upi"), mtd_spend=mtd, include_discovery=False) | |
| ranked_card = scored_card["held_ranked"] | |
| used_id = canonical_card_id(txn.get("card_id_used")) if txn.get("card_id_used") else None | |
| actual_rail = "upi" if txn.get("rail") == "upi" else "card" | |
| used_ranked = scored_upi["held_ranked"] if actual_rail == "upi" else ranked_card | |
| used = next((r for r in used_ranked if r["card_id"] == used_id), None) | |
| best_card = ranked_card[0] if ranked_card else None | |
| best_upi = scored_upi["held_ranked"][0] if scored_upi["held_ranked"] else None | |
| optimal = best_card | |
| optimal_rail = "card" | |
| if best_upi and best_upi["total_value_inr"] > (best_card["total_value_inr"] if best_card else 0): | |
| optimal = best_upi | |
| optimal_rail = "upi" | |
| lost = 0.0 | |
| was_optimal = True | |
| if used and optimal: | |
| lost = _r2(optimal["total_value_inr"] - used["total_value_inr"]) | |
| was_optimal = lost <= 0.01 | |
| total_lost += max(0.0, lost) | |
| # Only the physical card that was actually used consumes historical | |
| # issuer buckets. A counterfactual optimal card/rail must never rewrite | |
| # the real ledger used to score later purchases. | |
| used_card = CATALOGUE_BY_ID.get(canonical_card_id(used_id)) if used_id else None | |
| if used_card is not None: | |
| _accumulate_historical_spend(used_card, txn, history_state, statement_days) | |
| results[index] = { | |
| **txn, | |
| "used_card": used, | |
| "optimal_card": optimal, | |
| "was_optimal": was_optimal, | |
| "amount_lost": max(0.0, lost), | |
| "optimal_rail": optimal_rail, | |
| "rail_switch_helps": optimal_rail != actual_rail and not was_optimal, | |
| } | |
| return { | |
| "transactions": results, | |
| "total_lost_inr": _r2(total_lost), | |
| "count": len(results), | |
| "missed_count": sum(1 for r in results if r is not None and not r["was_optimal"]), | |
| } | |