""" 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 from typing import Dict, List, Optional from card_catalogue import Card, CATALOGUE_BY_ID, all_cards import math def _r2(x: float) -> float: """Round to 2 dp to exactly match JS Math.round(x*100)/100 (floor(y+0.5)) in engine.ts.""" return math.floor(x * 100 + 0.5) / 100 PERSONA_PHRASE = { "young_professional": "fits your everyday young-professional spending", "online_shopper": "matches how much you shop online", "traveller": "suits your travel spending", "frequent_traveller": "rewards your frequent travel", "food_lover": "is built for your dining habit", "value": "squeezes the most everyday value", "premium": "fits a premium lifestyle", "family": "works well for family spends", "bills_heavy": "pays you back on bills", "offers_hunter": "unlocks the most card offers", "amazon_loyal": "rewards your Amazon spending", "lifestyle": "fits your lifestyle spending", } def _persona_line(card_id: str, persona): card = CATALOGUE_BY_ID.get(card_id) fit = card.persona_fit if card else [] for p in (persona or []): if p in fit and p in PERSONA_PHRASE: return f"Personalised to you: {PERSONA_PHRASE[p]}." return None 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 direct = APPLY_URL.get(card_id) if direct: return direct card = CATALOGUE_BY_ID.get(card_id) import urllib.parse q = urllib.parse.quote(f"{card.name if card else ''} credit card apply {card.issuer if card else ''}") return f"https://www.google.com/search?q={q}" # RuPay credit-on-UPI rules (NPCI): rewards only accrue at/above the interchange # floor, and these categories earn nothing on the UPI rail. UPI_MIN_REWARD_AMOUNT = 2000 UPI_EXCLUDED_CATEGORIES = {"fuel", "rent", "wallet_load"} @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 @dataclass class CardScore: card_id: str card_name: str issuer: str network: str reward_value_inr: float # rewards earned on this txn (capped) instant_offer_inr: float # instant discount from live offers total_value_inr: float # reward + instant offer (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 def _offer_applies(off: Dict, card: Card) -> bool: cid = off.get("card_id") if cid: return cid == card.id iss = off.get("applies_to_issuer") or "" return iss == "" or iss == card.issuer 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 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 return best, note def score_transaction( held_card_ids: List[str], ctx: TxnContext, mtd_spend: Optional[Dict[str, Dict[str, float]]] = None, include_discovery: bool = True, 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 {} persona = persona or [] held_scores: List[CardScore] = [] all_scores: List[CardScore] = [] rail = getattr(ctx, "rail", "card") or "card" for card in all_cards(): held = card.id in held_card_ids # --- 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 < UPI_MIN_REWARD_AMOUNT: upi_block = f"UPI spends under ₹{UPI_MIN_REWARD_AMOUNT:,.0f} earn no rewards (issuers get no interchange below this)." 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 --- if upi_block: rate = 0.0 elif rail == "upi": rate = card.upi_effective_rate(ctx.category, ctx.brand_key, ctx.amount) else: rate = card.effective_rate(ctx.category, ctx.brand_key, ctx.amount) gross_units = ctx.amount / 100.0 * rate 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 = (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 = next((g for g in card.cap_groups if ctx.category in g.get("categories", [])), None) if brand_grp: cap_units = brand_grp["cap"] elif cat_grp: cap_units = cat_grp["cap"] else: cap_units = card.caps.get(ctx.category) if not upi_block and cap_units is not None and rate > 0: if brand_grp: spent = sum(mtd_spend.get(card.id, {}).get("~" + b, 0.0) for b in brand_grp["brands"]) elif cat_grp: spent = sum(mtd_spend.get(card.id, {}).get(c, 0.0) for c in cat_grp["categories"]) else: spent = mtd_spend.get(card.id, {}).get(ctx.category, 0.0) already_units = spent / 100.0 * rate remaining_units = max(0.0, cap_units - already_units) if gross_units > remaining_units: accel_units = remaining_units base_units_value = card.base_rate if ctx.category not in card.excluded_categories else 0.0 rupees_at_accel = remaining_units / rate * 100.0 if rate > 0 else 0.0 rupees_beyond = max(0.0, ctx.amount - rupees_at_accel) gross_units = accel_units + rupees_beyond / 100.0 * base_units_value 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 reward_inr = gross_units * card.point_value_inr # --- instant offers (don't apply on the UPI rail) --- if rail == "upi": offer_inr, offer_note = 0.0, None else: offer_inr, offer_note = _instant_offer_value(card, ctx) total = reward_inr + offer_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.") 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 offer_inr > 0 and offer_note: reasons.append(f"Live offer: {offer_note} (~₹{offer_inr:,.0f}).") # 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 = (ctx.amount / 100.0) * portal_rate portal_units = min(gross_units, card.portal_cap_units) if card.portal_cap_units is not None else gross_units portal_inr = portal_units * portal_pv portal_capped = card.portal_cap_units is not None and gross_units > card.portal_cap_units eff_portal_pct = (portal_inr / ctx.amount * 100.0) if ctx.amount else 0.0 if 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_value_inr=_r2(reward_inr), instant_offer_inr=_r2(offer_inr), total_value_inr=_r2(total), raw_total=total, effective_rate_pct=_r2(eff_pct), capped=capped, reasons=reasons, held=held, ) 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). _upi_ok = {c.id for c in all_cards() if getattr(c, "upi_eligible", False)} def _rank_key(s): demerit = 1 if (rail == "upi" and s.card_id not in _upi_ok) else 0 return (demerit, -s.raw_total) 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: 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), 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 analyse_transaction_history( held_card_ids: List[str], transactions: List[Dict], ) -> 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 = [] total_lost = 0.0 # month-to-date spend per CALENDAR MONTH, so monthly caps reset each month and a # year of spend cannot claim 12x a monthly cap. mtd_by_month: Dict[str, Dict[str, Dict[str, float]]] = {} for txn in transactions: _date = str(txn.get("date") or "") _mkey = _date[:7] if len(_date) >= 7 else "_" mtd = mtd_by_month.setdefault(_mkey, {}) 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, ) # 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 = txn.get("card_id_used") used = next((r for r in ranked_card 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) if used_id: amt = float(txn["amount"]) mtd.setdefault(used_id, {}).setdefault(txn["category"], 0.0) mtd[used_id][txn["category"]] += amt bk = txn.get("brand_key") if bk: mtd[used_id].setdefault("~" + bk, 0.0) mtd[used_id]["~" + bk] += amt results.append({ **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 == "upi" 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 not r["was_optimal"]), }