""" 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 @dataclass 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 @dataclass 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"]), }