from typing import Dict, Any, Optional, Tuple, List
import traceback
import json
import os
from datetime import date, datetime
import calendar
import gradio as gr
import plotly.graph_objects as go
import httpx
from config import (
APP_TITLE, APP_DESCRIPTION, THEME,
MCC_CATEGORIES, SAMPLE_USERS,
MERCHANTS_BY_CATEGORY)
from utils.api_client import RewardPilotClient
from utils.formatters import (
format_full_recommendation,
format_comparison_table,
format_analytics_metrics,
create_spending_chart,
create_rewards_pie_chart,
create_optimization_gauge,
create_trend_line_chart,
create_card_performance_chart)
from utils.llm_explainer import get_llm_explainer
from utils.gemini_explainer import get_gemini_explainer
import config
CARDS_FILE = os.path.join(os.path.dirname(__file__), "data", "cards.json")
def safe_get(data: Dict, key: str, default: Any = None) -> Any:
"""Safely get value from dictionary with fallback"""
try:
return data.get(key, default)
except:
return default
def normalize_recommendation_data(data: Dict) -> Dict:
"""
Normalize API response to ensure all required fields exist.
Handles both orchestrator format and mock data format.
"""
if data.get('mock_data'):
return {
'recommended_card': safe_get(data, 'recommended_card', 'Unknown Card'),
'rewards_earned': float(safe_get(data, 'rewards_earned', 0)),
'rewards_rate': safe_get(data, 'rewards_rate', 'N/A'),
'merchant': safe_get(data, 'merchant', 'Unknown Merchant'),
'category': safe_get(data, 'category', 'Unknown'),
'amount': float(safe_get(data, 'amount', 0)),
'annual_potential': float(safe_get(data, 'annual_potential', 0)),
'optimization_score': int(safe_get(data, 'optimization_score', 85)),
'reasoning': safe_get(data, 'reasoning', 'Optimal choice'),
'warnings': safe_get(data, 'warnings', []),
'alternatives': safe_get(data, 'alternatives', []),
'mock_data': True
}
recommended_card = safe_get(data, 'recommended_card', {})
if isinstance(recommended_card, dict):
card_name = safe_get(recommended_card, 'card_name', 'Unknown Card')
reward_amount = float(safe_get(recommended_card, 'reward_amount', 0))
reward_rate = float(safe_get(recommended_card, 'reward_rate', 0))
category = safe_get(recommended_card, 'category', 'Unknown')
reasoning = safe_get(recommended_card, 'reasoning', 'Optimal choice')
if reward_rate > 0:
rewards_rate_str = f"{reward_rate}x points"
else:
rewards_rate_str = "N/A"
else:
card_name = str(recommended_card) if recommended_card else 'Unknown Card'
reward_amount = float(safe_get(data, 'rewards_earned', 0))
reward_rate = 0
rewards_rate_str = safe_get(data, 'rewards_rate', 'N/A')
category = safe_get(data, 'category', 'Unknown')
reasoning = safe_get(data, 'reasoning', 'Optimal choice')
merchant = safe_get(data, 'merchant', 'Unknown Merchant')
amount = float(safe_get(data, 'amount_usd', safe_get(data, 'amount', 0)))
annual_potential = reward_amount * 12 if reward_amount > 0 else 0
alternatives = []
alt_cards = safe_get(data, 'alternative_cards', safe_get(data, 'alternatives', []))
for alt in alt_cards[:3]:
if isinstance(alt, dict):
alt_name = safe_get(alt, 'card_name', safe_get(alt, 'card', 'Unknown'))
alt_reward = float(safe_get(alt, 'reward_amount', safe_get(alt, 'rewards', 0)))
alt_rate = safe_get(alt, 'reward_rate', safe_get(alt, 'rate', 0))
if isinstance(alt_rate, (int, float)) and alt_rate > 0:
alt_rate_str = f"{alt_rate}x points"
else:
alt_rate_str = str(alt_rate) if alt_rate else "N/A"
alternatives.append({
'card': alt_name,
'rewards': alt_reward,
'rate': alt_rate_str
})
warnings = safe_get(data, 'warnings', [])
forecast_warning = safe_get(data, 'forecast_warning')
if forecast_warning and isinstance(forecast_warning, dict):
warning_msg = safe_get(forecast_warning, 'warning_message')
if warning_msg:
warnings.append(warning_msg)
normalized = {
'recommended_card': card_name,
'rewards_earned': round(reward_amount, 2),
'rewards_rate': rewards_rate_str,
'merchant': merchant,
'category': category,
'amount': amount,
'annual_potential': round(annual_potential, 2),
'optimization_score': int(safe_get(data, 'optimization_score', 75)),
'reasoning': reasoning,
'warnings': warnings,
'alternatives': alternatives,
'mock_data': safe_get(data, 'mock_data', False)
}
return normalized
def create_loading_state():
"""Create loading indicator message"""
return "⏳ **Loading...** Please wait while we fetch your recommendation.", None
def load_card_database() -> dict:
"""Load card database from local cards.json"""
try:
with open(CARDS_FILE, 'r') as f:
cards = json.load(f)
print(f"✅ Loaded {len(cards)} cards from database")
return cards
except FileNotFoundError:
print(f"⚠️ cards.json not found at {CARDS_FILE}")
return {}
except json.JSONDecodeError as e:
print(f"❌ Error parsing cards.json: {e}")
return {}
CARD_DATABASE = load_card_database()
def get_card_details(card_id: str, mcc: str = None) -> dict:
"""
Get card details from database
Args:
card_id: Card identifier (e.g., "c_citi_custom_cash")
mcc: Optional MCC code to get specific reward rate
Returns:
dict: Card details including name, reward rate, caps, etc.
"""
if card_id not in CARD_DATABASE:
print(f"⚠️ Card {card_id} not found in database, using fallback")
return {
"name": card_id.replace("c_", "").replace("_", " ").title(),
"issuer": "Unknown",
"reward_rate": 1.0,
"annual_fee": 0,
"spending_caps": {},
"benefits": []
}
card = CARD_DATABASE[card_id]
reward_rate = 1.0
if mcc and "reward_structure" in card:
reward_structure = card["reward_structure"]
if mcc in reward_structure:
reward_rate = reward_structure[mcc]
else:
try:
mcc_int = int(mcc)
for key, rate in reward_structure.items():
if "-" in str(key):
start, end = str(key).split("-")
if int(start) <= mcc_int <= int(end):
reward_rate = rate
break
except (ValueError, AttributeError):
pass
if reward_rate == 1.0 and "default" in reward_structure:
reward_rate = reward_structure["default"]
spending_caps = card.get("spending_caps", {})
cap_info = {}
if "monthly_bonus" in spending_caps:
cap_info = {
"type": "monthly",
"limit": spending_caps["monthly_bonus"],
"display": f"${spending_caps['monthly_bonus']}/month"
}
elif "quarterly_bonus" in spending_caps:
cap_info = {
"type": "quarterly",
"limit": spending_caps["quarterly_bonus"],
"display": f"${spending_caps['quarterly_bonus']}/quarter"
}
elif "annual_bonus" in spending_caps:
cap_info = {
"type": "annual",
"limit": spending_caps["annual_bonus"],
"display": f"${spending_caps['annual_bonus']}/year"
}
return {
"name": card.get("name", "Unknown Card"),
"issuer": card.get("issuer", "Unknown"),
"reward_rate": reward_rate,
"annual_fee": card.get("annual_fee", 0),
"spending_caps": cap_info,
"benefits": card.get("benefits", [])
}
def get_recommendation_with_agent(user_id, merchant, category, amount):
import time
# Stage 1
yield """
⏳
AI Agent is thinking...
⏳
Analyzing your wallet...
⏳
Comparing reward rates...
⏳
Calculating optimal strategy...
""", None
time.sleep(0.8)
try:
transaction = {
"user_id": user_id,
"merchant": merchant,
"category": category,
"mcc": MCC_CATEGORIES.get(category, "5999"),
"amount_usd": float(amount)
}
print("=" * 80)
print(f"🚀 REQUEST: {config.ORCHESTRATOR_URL}/recommend")
print(f"PAYLOAD: {json.dumps(transaction, indent=2)}")
# Stage 2
yield """
✅
AI Agent is thinking
⏳
Analyzing your wallet...
⏳
Comparing reward rates...
⏳
Calculating optimal strategy...
""", None
time.sleep(0.6)
response = httpx.post(
f"{config.ORCHESTRATOR_URL}/recommend",
json=transaction,
timeout=60.0
)
# Stage 3
yield """
✅
AI Agent is thinking
✅
Analyzing your wallet
⏳
Comparing reward rates...
⏳
Calculating optimal strategy...
""", None
time.sleep(0.6)
print(f"📥 STATUS: {response.status_code}")
print(f"📦 RESPONSE: {response.text[:2000]}")
print("=" * 80)
if response.status_code != 200:
yield f"❌ Error: API returned status {response.status_code}", None
return
result = response.json()
# Stage 4
yield """
✅
AI Agent is thinking
✅
Analyzing your wallet
✅
Comparing reward rates
⏳
Calculating optimal strategy...
""", None
time.sleep(0.5)
if not isinstance(result, dict):
yield f"❌ Invalid response type: {type(result)}", None
return
print(f"🔍 KEYS: {list(result.keys())}")
card_id = result.get('recommended_card', 'Unknown')
rewards_earned = float(result.get('rewards_earned', 0))
rewards_rate = result.get('rewards_rate', 'N/A')
confidence = float(result.get('confidence', 0))
reasoning = result.get('reasoning', 'No reasoning provided')
alternatives = result.get('alternative_options', [])
warnings = result.get('warnings', [])
card_name_map = {
'c_citi_custom_cash': 'Citi Custom Cash',
'c_amex_gold': 'American Express Gold',
'c_chase_sapphire_reserve': 'Chase Sapphire Reserve',
'c_chase_freedom_unlimited': 'Chase Freedom Unlimited',
'c_chase_sapphire_preferred': 'Chase Sapphire Preferred',
'c_capital_one_venture': 'Capital One Venture',
'c_discover_it': 'Discover it',
'c_wells_fargo_active_cash': 'Wells Fargo Active Cash'
}
card_name = card_name_map.get(card_id, card_id.replace('c_', '').replace('_', ' ').title())
transaction_mcc = result.get('mcc', MCC_CATEGORIES.get(category, "5999"))
card_details_from_db = get_card_details(card_id, transaction_mcc)
card_details = result.get('card_details', {})
if not card_details or not card_details.get('reward_rate'):
reward_structure = CARD_DATABASE.get(card_id, {}).get('reward_structure', {})
if transaction_mcc in reward_structure:
reward_rate_value = reward_structure[transaction_mcc]
else:
reward_rate_value = reward_structure.get('default', 1.0)
spending_caps_db = CARD_DATABASE.get(card_id, {}).get('spending_caps', {})
card_details = {
'reward_rate': reward_rate_value,
'monthly_cap': spending_caps_db.get('monthly_bonus'),
'annual_cap': spending_caps_db.get('annual_bonus'),
'quarterly_cap': spending_caps_db.get('quarterly_bonus'),
'base_rate': reward_structure.get('default', 1.0),
'annual_fee': CARD_DATABASE.get(card_id, {}).get('annual_fee', 0),
'cap_type': 'monthly' if 'monthly_bonus' in spending_caps_db else
'annual' if 'annual_bonus' in spending_caps_db else
'quarterly' if 'quarterly_bonus' in spending_caps_db else 'none'
}
print(f"✅ Using cards.json details for {card_id}")
reward_rate_value = card_details.get('reward_rate', 1.0)
monthly_cap = card_details.get('monthly_cap', None)
annual_cap = card_details.get('annual_cap', None)
base_rate = card_details.get('base_rate', 1.0)
annual_fee = card_details.get('annual_fee', 0)
print(f"✅ CARD DETAILS: {reward_rate_value}%, cap={monthly_cap or annual_cap}, fee=${annual_fee}")
amount_float = float(amount)
frequency_map = {
'Groceries': 52,
'Restaurants': 52,
'Gas Stations': 52,
'Fast Food': 52,
'Airlines': 4,
'Hotels': 12,
'Online Shopping': 24,
'Entertainment': 24,
}
frequency = frequency_map.get(category, 26)
frequency_label = {
52: 'weekly',
26: 'bi-weekly',
24: 'bi-weekly',
12: 'monthly',
4: 'quarterly'
}.get(frequency, f'{frequency}x per year')
annual_spend = amount_float * frequency
if monthly_cap:
monthly_cap_annual = monthly_cap * 12
if annual_spend <= monthly_cap_annual:
high_rate_spend = annual_spend
low_rate_spend = 0
else:
high_rate_spend = monthly_cap_annual
low_rate_spend = annual_spend - monthly_cap_annual
high_rate_rewards = high_rate_spend * (reward_rate_value / 100)
low_rate_rewards = low_rate_spend * (base_rate / 100)
total_rewards = high_rate_rewards + low_rate_rewards
calc_table = f"""
| Spending Tier | Annual Amount | Rate | Rewards |
|---------------|---------------|------|---------|
| First ${monthly_cap}/month | ${high_rate_spend:.2f} | {reward_rate_value}% | ${high_rate_rewards:.2f} |
| Remaining spend | ${low_rate_spend:.2f} | {base_rate}% | ${low_rate_rewards:.2f} |
| **Subtotal** | **${annual_spend:.2f}** | - | **${total_rewards:.2f}** |
| Annual fee | - | - | -${annual_fee:.2f} |
| **Net Rewards** | - | - | **${total_rewards - annual_fee:.2f}** |
"""
elif annual_cap:
if annual_spend <= annual_cap:
high_rate_spend = annual_spend
low_rate_spend = 0
else:
high_rate_spend = annual_cap
low_rate_spend = annual_spend - annual_cap
high_rate_rewards = high_rate_spend * (reward_rate_value / 100)
low_rate_rewards = low_rate_spend * (base_rate / 100)
total_rewards = high_rate_rewards + low_rate_rewards
calc_table = f"""
| Spending Tier | Annual Amount | Rate | Rewards |
|---------------|---------------|------|---------|
| Up to ${annual_cap:,.0f}/year | ${high_rate_spend:.2f} | {reward_rate_value}% | ${high_rate_rewards:.2f} |
| Above cap | ${low_rate_spend:.2f} | {base_rate}% | ${low_rate_rewards:.2f} |
| **Subtotal** | **${annual_spend:.2f}** | - | **${total_rewards:.2f}** |
| Annual fee | - | - | -${annual_fee:.2f} |
| **Net Rewards** | - | - | **${total_rewards - annual_fee:.2f}** |
"""
else:
total_rewards = annual_spend * (reward_rate_value / 100)
calc_table = f"""
| Spending Tier | Annual Amount | Rate | Rewards |
|---------------|---------------|------|---------|
| All spending | ${annual_spend:.2f} | {reward_rate_value}% | ${total_rewards:.2f} |
| Annual fee | - | - | -${annual_fee:.2f} |
| **Net Rewards** | - | - | **${total_rewards - annual_fee:.2f}** |
"""
baseline_rewards = annual_spend * 0.01
net_rewards = total_rewards - annual_fee
net_benefit = net_rewards - baseline_rewards
comparison_text = f"""
**With {card_name}:**
- Earnings: ${total_rewards:.2f}
- Annual fee: -${annual_fee:.2f}
- **Net total: ${net_rewards:.2f}/year**
**With Baseline 1% Card:**
- All spending at 1%: ${baseline_rewards:.2f}/year
**Net Benefit: ${net_benefit:+.2f}/year** {"🎉" if net_benefit > 0 else "⚠️"}
"""
max_possible_rewards = annual_spend * 0.06
if max_possible_rewards > 0:
performance_ratio = (net_rewards / max_possible_rewards) * 100
if net_rewards > baseline_rewards:
improvement = (net_rewards - baseline_rewards) / baseline_rewards
baseline_bonus = min(improvement * 20, 20)
else:
baseline_bonus = -10
optimization_score = int(min(performance_ratio + baseline_bonus, 100))
else:
optimization_score = 0
score_breakdown = {
'reward_rate': min(30, int(optimization_score * 0.30)),
'cap_availability': min(25, int(optimization_score * 0.25)),
'annual_fee': min(20, int(optimization_score * 0.20)),
'category_match': min(20, int(optimization_score * 0.20)),
'penalties': max(-5, int((optimization_score - 100) * 0.05))
}
score_details = f"""
**Score Components:**
- {"✅" if score_breakdown['reward_rate'] > 20 else "⚠️"} Reward rate: **+{score_breakdown['reward_rate']} points**
- {"✅" if score_breakdown['cap_availability'] > 15 else "⚠️"} Cap availability: **+{score_breakdown['cap_availability']} points**
- {"✅" if score_breakdown['annual_fee'] > 15 else "⚠️"} Annual fee value: **+{score_breakdown['annual_fee']} points**
- {"✅" if score_breakdown['category_match'] > 15 else "⚠️"} Category match: **+{score_breakdown['category_match']} points**
- {"⚠️" if score_breakdown['penalties'] < 0 else "✅"} Limitations: **{score_breakdown['penalties']} points**
**Total: {optimization_score}/100**
**Score Ranges:**
- 90-100: Optimal choice ✅
- 80-89: Great choice 👍
- 70-79: Good choice 👌
- 60-69: Acceptable ⚠️
- <60: Suboptimal ❌
"""
def format_reasoning(text):
"""Format reasoning text into clean bullet points"""
if text.strip().startswith(('-', '•', '*', '1.', '2.')):
return text
sentences = text.replace('\n', ' ').split('. ')
bullets = []
for sentence in sentences[:4]:
sentence = sentence.strip()
if sentence and len(sentence) > 20:
if not sentence.endswith('.'):
sentence += '.'
bullets.append(f"- {sentence}")
return '\n'.join(bullets) if bullets else f"- {text}"
reasoning_bullets = format_reasoning(reasoning)
# Format the output - CRITICAL: No indentation on f-string!
output = f"""## 🎯 Recommended: **{card_name}**
| Metric | Value |
|--------|-------|
| 💰 **Rewards Earned** | ${rewards_earned:.2f} ({rewards_rate}) |
| 📊 **Confidence** | {confidence*100:.0f}% |
| 📈 **Annual Potential** | ${net_benefit:.2f}/year |
| ⭐ **Optimization Score** | {optimization_score}/100 |
---
### 🧠 Why This Card?
{reasoning_bullets}
---
"""
# Alternatives
if alternatives:
output += "\n### 🔄 Alternative Options\n\n"
output += "| Card | Rewards | Why? |\n"
output += "|------|---------|------|\n"
for alt in alternatives[:3]:
alt_card_id = alt.get('card', '')
alt_card_name = card_name_map.get(alt_card_id, alt_card_id.replace('c_', '').replace('_', ' ').title())
alt_reason = alt.get('reason', 'Good alternative')
alt_reward = alt.get('reward_amount', rewards_earned * 0.8)
alt_reason_short = alt_reason.split('.')[0].strip()
if not alt_reason_short.endswith('.'):
alt_reason_short += '.'
output += f"| {alt_card_name} | ${alt_reward:.2f} | {alt_reason_short} |\n"
output += "\n---\n"
# Warnings
if warnings:
output += "\n### ⚠️ Alerts\n\n"
for warning in warnings:
output += f"- {warning}\n"
output += "\n---\n"
# Calculation details (collapsible)
output += f"""
📊 Annual Impact Calculation (Click to expand)
**Assumptions:**
- Transaction: ${amount_float:.2f} at {merchant} ({category})
- Frequency: {frequency_label} → ${annual_spend:.2f}/year
**Rewards Breakdown:**
{calc_table}
**vs. Baseline (1% card):** ${baseline_rewards:.2f}/year
**Net Benefit:** ${net_benefit:+.2f}/year {"🎉" if net_benefit > 0 else "⚠️"}
**Card Details:** {reward_rate_value}% on {category} | Cap: {"$" + str(monthly_cap or annual_cap) if (monthly_cap or annual_cap) else "None"} | Fee: ${annual_fee}
"""
chart = create_agent_recommendation_chart_enhanced(result)
yield output, chart
print("=" * 80)
print("📤 FINAL OUTPUT (first 500 chars):")
print(output[:500])
print("=" * 80)
except Exception as e:
print(f"❌ ERROR: {traceback.format_exc()}")
yield f"❌ **Error:** {str(e)}", None
def create_agent_recommendation_chart_enhanced(result: Dict) -> go.Figure:
try:
rec_name_map = {
'c_citi_custom_cash': 'Citi Custom Cash',
'c_amex_gold': 'Amex Gold',
'c_chase_sapphire_reserve': 'Sapphire Reserve',
'c_chase_freedom_unlimited': 'Freedom Unlimited'
}
rec_id = result.get('recommended_card', '')
rec_name = rec_name_map.get(rec_id, rec_id)
rec_reward = float(result.get('rewards_earned', 0))
cards = [rec_name]
rewards = [rec_reward]
colors = ['#667eea']
alternatives = result.get('alternative_options', [])
for alt in alternatives[:3]:
alt_id = alt.get('card', '')
alt_name = rec_name_map.get(alt_id, alt_id)
alt_reward = rec_reward * 0.8
cards.append(alt_name)
rewards.append(alt_reward)
colors.append('#cbd5e0')
fig = go.Figure(data=[
go.Bar(
x=cards,
y=rewards,
marker=dict(color=colors, line=dict(color='white', width=2)),
text=[f'${r:.2f}' for r in rewards],
textposition='outside'
)
])
fig.update_layout(
title='🎯 Card Comparison',
xaxis_title='Credit Card',
yaxis_title='Rewards ($)',
template='plotly_white',
height=400,
showlegend=False
)
return fig
except Exception as e:
print(f"Chart error: {e}")
fig = go.Figure()
fig.add_annotation(text="Chart unavailable", xref="paper", yref="paper", x=0.5, y=0.5, showarrow=False)
fig.update_layout(height=400, template='plotly_white')
return fig
client = RewardPilotClient(config.ORCHESTRATOR_URL)
llm = get_llm_explainer()
gemini = get_gemini_explainer()
def get_recommendation(
user_id: str,
merchant: str,
category: str,
amount: float,
use_custom_mcc: bool,
custom_mcc: str,
transaction_date: Optional[str]) -> tuple:
"""Get card recommendation and format response"""
if not user_id or not merchant or amount <= 0:
return (
"❌ **Error:** Please fill in all required fields.",
None,
None,
)
if use_custom_mcc and custom_mcc:
mcc = custom_mcc
else:
mcc = MCC_CATEGORIES.get(category, "5999")
if not transaction_date:
transaction_date = str(date.today())
response: Dict[str, Any] = client.get_recommendation_sync(
user_id=user_id,
merchant=merchant,
mcc=mcc,
amount_usd=amount,
transaction_date=transaction_date,
)
formatted_text = format_full_recommendation(response)
comparison_table: Optional[str]
stats: Optional[str]
if not response.get("error"):
recommended = response.get("recommended_card", {}) or {}
alternatives: List[Dict[str, Any]] = response.get("alternative_cards", []) or []
all_cards = [c for c in ([recommended] + alternatives) if c]
comparison_table = format_comparison_table(all_cards) if all_cards else None
total_analyzed = response.get("total_cards_analyzed", len(all_cards))
best_reward = (recommended.get("reward_amount") or 0.0)
services_used = response.get("services_used", [])
stats = f"""**Cards Analyzed:** {total_analyzed}
**Best Reward:** ${best_reward:.2f}
**Services Used:** {', '.join(services_used)}""".strip()
else:
comparison_table = None
stats = None
return formatted_text, comparison_table, stats
def get_recommendation_with_ai(user_id, merchant, category, amount):
"""Get card recommendation with LLM-powered explanation"""
if not merchant or not merchant.strip():
return "❌ Please enter a merchant name.", None
if amount <= 0:
return "❌ Please enter a valid amount greater than $0.", None
yield "⏳ **Loading recommendation...** Analyzing your cards and transaction...", None
try:
result = client.get_recommendation(
user_id=user_id,
merchant=merchant,
category=category,
amount=float(amount),
mcc=None
)
if not result.get('success'):
error_msg = result.get('error', 'Unknown error')
yield f"❌ Error: {error_msg}", None
return
data = normalize_recommendation_data(result.get('data', {}))
ai_explanation = ""
if config.USE_GEMINI and gemini.enabled:
try:
ai_explanation = gemini.explain_recommendation(
card=data['recommended_card'],
rewards=data['rewards_earned'],
rewards_rate=data['rewards_rate'],
merchant=merchant,
category=category,
amount=float(amount),
warnings=data['warnings'] if data['warnings'] else None,
annual_potential=data['annual_potential'],
alternatives=data['alternatives']
)
ai_explanation = f"🤖 **Powered by Google Gemini**\n\n{ai_explanation}"
except Exception as e:
print(f"Gemini explanation failed: {e}")
ai_explanation = ""
output = f"""
## 🎯 Recommendation for ${amount:.2f} at {merchant}
### 💳 Best Card: **{data['recommended_card']}**
**Rewards Earned:** ${data['rewards_earned']:.2f} ({data['rewards_rate']})
"""
if data.get('mock_data'):
output += """
> ⚠️ **Demo Mode:** Using sample data. Connect to orchestrator for real recommendations.
"""
if ai_explanation:
output += f"""
### 🤖 AI Insight
{ai_explanation}
---
"""
output += f"""
### 📊 Breakdown
- **Category:** {data['category']}
- **Merchant:** {data['merchant']}
- **Reasoning:** {data['reasoning']}
- **Annual Potential:** ${data['annual_potential']:.2f}
- **Optimization Score:** {data['optimization_score']}/100
"""
if data['warnings']:
output += "\n\n### ⚠️ Important Warnings\n\n"
for warning in data['warnings']:
output += f"- {warning}\n"
if data['alternatives']:
output += "\n\n### 🔄 Alternative Options\n\n"
for alt in data['alternatives']:
output += f"- **{alt['card']}:** ${alt['rewards']:.2f} ({alt['rate']})\n"
chart = create_rewards_comparison_chart(data)
yield output, chart
except Exception as e:
error_details = traceback.format_exc()
print(f"Recommendation error: {error_details}")
yield f"❌ Error: {str(e)}\n\nPlease check your API connection or try again.", None
def create_rewards_comparison_chart(data: Dict) -> go.Figure:
"""Create rewards comparison chart with proper error handling"""
try:
cards = [data['recommended_card']]
rewards = [data['rewards_earned']]
colors = ['#667eea']
for alt in data.get('alternatives', [])[:3]:
cards.append(alt['card'])
rewards.append(float(alt['rewards']))
colors.append('#a0aec0')
if not cards or all(r == 0 for r in rewards):
fig = go.Figure()
fig.add_annotation(
text="No rewards data available for comparison",
xref="paper", yref="paper",
x=0.5, y=0.5, showarrow=False,
font=dict(size=14, color="#666")
)
fig.update_layout(
height=400,
template='plotly_white'
)
return fig
fig = go.Figure(data=[
go.Bar(
x=cards,
y=rewards,
marker=dict(
color=colors,
line=dict(color='white', width=2)
),
text=[f'${r:.2f}' for r in rewards],
textposition='outside',
hovertemplate='%{x}
Rewards: $%{y:.2f}'
)
])
fig.update_layout(
title={
'text': 'Rewards Comparison',
'x': 0.5,
'xanchor': 'center'
},
xaxis_title='Credit Card',
yaxis_title='Rewards Earned ($)',
template='plotly_white',
height=400,
showlegend=False,
margin=dict(t=60, b=50, l=50, r=50),
hovermode='x'
)
return fig
except Exception as e:
print(f"Chart creation error: {e}")
print(traceback.format_exc())
fig = go.Figure()
fig.add_annotation(
text=f"Error creating chart",
xref="paper", yref="paper",
x=0.5, y=0.5, showarrow=False,
font=dict(size=14, color="red")
)
fig.update_layout(height=400, template='plotly_white')
return fig
def get_analytics_with_insights(user_id):
"""Get analytics with LLM-generated insights"""
try:
result = client.get_user_analytics(user_id)
if not result.get('success'):
return f"❌ Error: {result.get('error', 'Unknown error')}", None, None, None
data = result['data']
ai_insights = ""
if config.LLM_ENABLED:
try:
ai_insights = llm.generate_spending_insights(
user_id=user_id,
total_spending=data['total_spending'],
total_rewards=data['total_rewards'],
optimization_score=data['optimization_score'],
top_categories=data.get('category_breakdown', []),
recommendations_count=data.get('optimized_count', 0)
)
except Exception as e:
print(f"AI insights generation failed: {e}")
ai_insights = ""
metrics = f"""
## 📊 Your Rewards Analytics
### Key Metrics
- **💰 Total Rewards:** ${data['total_rewards']:.2f}
- **📈 Potential Savings:** ${data['potential_savings']:.2f}/year
- **⭐ Optimization Score:** {data['optimization_score']}/100
- **✅ Optimized Transactions:** {data.get('optimized_count', 0)}
"""
if ai_insights:
metrics += f"""
### 🤖 Personalized Insights
{ai_insights}
---
"""
spending_chart = create_spending_chart(data)
rewards_chart = create_rewards_distribution_chart(data)
optimization_chart = create_optimization_gauge(data['optimization_score'])
return metrics, spending_chart, rewards_chart, optimization_chart
except Exception as e:
return f"❌ Error: {str(e)}", None, None, None
EXAMPLES = [
["u_alice", "Groceries", "Whole Foods", 125.50, False, "", "2025-01-15"],
["u_bob", "Restaurants", "Olive Garden", 65.75, False, "", "2025-01-15"],
["u_charlie", "Airlines", "United Airlines", 450.00, False, "", "2025-01-15"],
["u_alice", "Fast Food", "Starbucks", 15.75, False, "", ""],
["u_bob", "Gas Stations", "Shell", 45.00, False, "", ""],
]
def create_empty_chart(message: str) -> go.Figure:
"""Helper to create empty chart with message"""
fig = go.Figure()
fig.add_annotation(
text=message,
xref="paper", yref="paper",
x=0.5, y=0.5, showarrow=False,
font=dict(size=14, color="#666")
)
fig.update_layout(height=400, template='plotly_white')
return fig
# ===================== NEW: FORMAT CURRENT MONTH SUMMARY =====================
def format_current_month_summary(analytics_data):
"""Format current month warnings with clear styling (for Analytics tab)"""
warnings = []
# Check for spending cap warnings
for card in analytics_data.get('card_usage', []):
if card.get('cap_percentage', 0) > 90:
warnings.append(
f"⚠️ {card['name']}: "
f"${card['current_spend']:.0f} / ${card['cap']:.0f} "
f"({card['cap_percentage']:.0f}% used)"
)
# Calculate end-of-month projection
days_elapsed = datetime.now().day
days_in_month = calendar.monthrange(datetime.now().year, datetime.now().month)[1]
projection_ratio = days_in_month / days_elapsed if days_elapsed > 0 else 1
projected_spending = analytics_data.get('total_spending', 0) * projection_ratio
projected_rewards = analytics_data.get('total_rewards', 0) * projection_ratio
warnings_html = "
".join(warnings) if warnings else "✅ No warnings - you're on track!"
return f"""
⚠️ This Month's Status (as of {datetime.now().strftime('%B %d')})
Month-End Projection:
- Estimated Total Spending: ${projected_spending:.2f}
- Estimated Total Rewards: ${projected_rewards:.2f}
Spending Cap Alerts:
{warnings_html}
💡 These are estimates based on your current month's activity.
For detailed future predictions, visit the Forecast tab.
"""
def update_analytics_with_charts(user_id: str):
"""Fetch and format analytics with charts for selected user"""
try:
result = client.get_user_analytics(user_id)
print("=" * 60)
print(f"DEBUG: Analytics for {user_id}")
print(f"Success: {result.get('success')}")
if result.get('data'):
print(f"Data keys: {result['data'].keys()}")
print(f"Total spending: {result['data'].get('total_spending')}")
print(f"Total rewards: {result['data'].get('total_rewards')}")
print("=" * 60)
if not result.get('success'):
error_msg = result.get('error', 'Unknown error')
empty_fig = create_empty_chart(f"Error: {error_msg}")
return (
f"❌ Error: {error_msg}
",
empty_fig, empty_fig, empty_fig, empty_fig, empty_fig,
"Error loading data",
"Error loading data",
f"Error: {error_msg}
", # Changed from forecast_md
f"*Error: {error_msg}*"
)
analytics_data = result.get('data', {})
if not analytics_data:
empty_fig = create_empty_chart("No analytics data available")
return (
"No data available
",
empty_fig, empty_fig, empty_fig, empty_fig, empty_fig,
"No data",
"No data",
"No data available
", # Changed
"*No data available*"
)
metrics_html, table_md, insights_md, _ = format_analytics_metrics(analytics_data)
# Generate current month summary (NEW)
current_month_html = format_current_month_summary(analytics_data)
spending_fig = create_spending_chart(analytics_data)
pie_fig = create_rewards_pie_chart(analytics_data)
gauge_fig = create_optimization_gauge(analytics_data)
trend_fig = create_trend_line_chart(analytics_data)
performance_fig = create_card_performance_chart(analytics_data)
status = f"*Analytics updated for {user_id} at {datetime.now().strftime('%I:%M %p')}*"
return (
metrics_html,
spending_fig,
gauge_fig,
pie_fig,
performance_fig,
trend_fig,
table_md,
insights_md,
current_month_html, # Changed from forecast_md
status
)
except Exception as e:
error_details = traceback.format_exc()
error_msg = f"❌ Error loading analytics: {str(e)}"
print(error_msg)
print(error_details)
empty_fig = create_empty_chart("Error loading chart")
return (
f"{error_msg}
",
empty_fig, empty_fig, empty_fig, empty_fig, empty_fig,
"Error loading table",
"Error loading insights",
f"{error_msg}
", # Changed
f"*{error_msg}*"
)
def _toggle_custom_mcc(use_custom: bool):
return gr.update(visible=use_custom, value="")
# ===================== NEW FUNCTION FOR SMART WALLET =====================
def load_user_wallet(user_id: str):
"""Load and display user's credit card wallet"""
try:
# This would call your Smart Wallet API
# For now, using mock data structure
wallet_data = {
'u_alice': [
{'name': 'Amex Gold', 'issuer': 'American Express', 'status': 'Active', 'limit': '$25,000'},
{'name': 'Chase Sapphire Reserve', 'issuer': 'Chase', 'status': 'Active', 'limit': '$30,000'},
{'name': 'Citi Custom Cash', 'issuer': 'Citibank', 'status': 'Active', 'limit': '$15,000'},
],
'u_bob': [
{'name': 'Chase Freedom Unlimited', 'issuer': 'Chase', 'status': 'Active', 'limit': '$20,000'},
{'name': 'Discover it', 'issuer': 'Discover', 'status': 'Active', 'limit': '$12,000'},
],
'u_charlie': [
{'name': 'Capital One Venture', 'issuer': 'Capital One', 'status': 'Active', 'limit': '$35,000'},
{'name': 'Wells Fargo Active Cash', 'issuer': 'Wells Fargo', 'status': 'Active', 'limit': '$18,000'},
]
}
cards = wallet_data.get(user_id, [])
if not cards:
return "No cards found in wallet", create_empty_chart("No cards in wallet")
output = f"## 💳 Your Credit Card Wallet ({len(cards)} cards)\n\n"
for card in cards:
output += f"""
### {card['name']}
- **Issuer:** {card['issuer']}
- **Status:** {card['status']}
- **Credit Limit:** {card['limit']}
---
"""
# Create simple chart showing card limits
fig = go.Figure(data=[
go.Bar(
x=[c['name'] for c in cards],
y=[int(c['limit'].replace('$', '').replace(',', '')) for c in cards],
marker=dict(color='#667eea'),
text=[c['limit'] for c in cards],
textposition='outside'
)
])
fig.update_layout(
title='Credit Limits by Card',
xaxis_title='Card',
yaxis_title='Credit Limit ($)',
template='plotly_white',
height=400
)
return output, fig
except Exception as e:
return f"❌ Error loading wallet: {str(e)}", create_empty_chart("Error")
# ===================== NEW FUNCTION FOR FORECAST =====================
def load_user_forecast(user_id: str):
"""Load and display spending forecast (comprehensive version for Forecast tab)"""
try:
# Mock forecast data - replace with actual API call
forecast_data = {
'next_month_spending': 3250.50,
'predicted_rewards': 127.50,
'confidence': 0.92,
'top_categories': [
{'category': 'Groceries', 'predicted': 850.00, 'confidence': 0.92, 'emoji': '🛒'},
{'category': 'Restaurants', 'predicted': 650.00, 'confidence': 0.88, 'emoji': '🍽️'},
{'category': 'Gas', 'predicted': 450.00, 'confidence': 0.85, 'emoji': '⛽'},
],
'recommendations': [
"Use Amex Gold for groceries (4x points)",
"Approaching Citi Custom Cash $500 cap",
"Travel spending predicted to increase"
],
'optimization_potential': 45.50
}
confidence = forecast_data.get('confidence', 0.85)
confidence_badge = "High" if confidence > 0.9 else "Medium" if confidence > 0.75 else "Low"
confidence_color = '#4caf50' if confidence > 0.9 else '#ff9800' if confidence > 0.75 else '#f44336'
# Compact output
output = f"""
## 🔮 Next Month Forecast
**Confidence:** {confidence*100:.0f}% ({confidence_badge})
### 📊 Summary
| Metric | Amount |
|--------|--------|
| 💰 **Total Spending** | ${forecast_data['next_month_spending']:.2f} |
| 🎁 **Expected Rewards** | ${forecast_data['predicted_rewards']:.2f} |
| 📈 **Extra with Optimization** | +${forecast_data['optimization_potential']:.2f} |
---
### 📊 Top Categories
| Category | Predicted | Confidence |
|----------|-----------|------------|
"""
for cat in forecast_data['top_categories']:
output += f"| {cat['emoji']} {cat['category']} | ${cat['predicted']:.2f} | {cat['confidence']*100:.0f}% |\n"
output += "\n---\n\n### 💡 Action Items\n\n"
for i, rec in enumerate(forecast_data['recommendations'], 1):
output += f"{i}. {rec}\n"
output += """
---
**💡 Tip:** Check the Analytics tab to see your current spending patterns and optimization opportunities.
"""
# Create forecast chart
fig = go.Figure()
categories = [c['category'] for c in forecast_data['top_categories']]
amounts = [c['predicted'] for c in forecast_data['top_categories']]
confidences = [c['confidence'] for c in forecast_data['top_categories']]
# Color bars based on confidence
colors = ['#4caf50' if c > 0.9 else '#ff9800' if c > 0.8 else '#f44336' for c in confidences]
fig.add_trace(go.Bar(
x=categories,
y=amounts,
marker=dict(color=colors),
text=[f'${a:.0f}
{c*100:.0f}% conf.' for a, c in zip(amounts, confidences)],
textposition='outside',
hovertemplate='%{x}
Predicted: $%{y:.2f}'
))
fig.update_layout(
title='Predicted Spending by Category (Next Month)',
xaxis_title='Category',
yaxis_title='Predicted Amount ($)',
template='plotly_white',
height=400,
showlegend=False
)
return output, fig
except Exception as e:
return f"❌ Error loading forecast: {str(e)}", create_empty_chart("Error")
# ===================== MAIN GRADIO APP =====================
with gr.Blocks(
theme=THEME if isinstance(THEME, gr.themes.ThemeClass) else gr.themes.Soft(),
title=APP_TITLE,
css="""
.gradio-container {
max-width: 1200px !important;
}
/* ===== HERO SECTION STYLES ===== */
.hero-section {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
padding: 50px 40px;
border-radius: 20px;
color: white;
margin-bottom: 35px;
box-shadow: 0 15px 40px rgba(102, 126, 234, 0.4);
position: relative;
overflow: hidden;
}
.hero-section::before {
content: '';
position: absolute;
top: -50%;
right: -50%;
width: 200%;
height: 200%;
background: radial-gradient(circle, rgba(255,255,255,0.1) 0%, transparent 70%);
animation: pulse 4s ease-in-out infinite;
}
@keyframes pulse {
0%, 100% { transform: scale(1); opacity: 0.5; }
50% { transform: scale(1.1); opacity: 0.8; }
}
.hero-title {
font-size: 42px;
font-weight: 800;
margin-bottom: 15px;
text-align: center;
text-shadow: 0 2px 10px rgba(0,0,0,0.2);
position: relative;
z-index: 1;
}
.hero-subtitle {
font-size: 22px;
font-weight: 400;
margin-bottom: 35px;
text-align: center;
opacity: 0.95;
line-height: 1.6;
position: relative;
z-index: 1;
}
.impact-stats {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(200px, 1fr));
gap: 25px;
margin-top: 35px;
position: relative;
z-index: 1;
}
.stat-card {
background: rgba(255, 255, 255, 0.2);
backdrop-filter: blur(10px);
padding: 25px;
border-radius: 15px;
border: 2px solid rgba(255, 255, 255, 0.3);
text-align: center;
transition: all 0.3s ease;
}
.stat-card:hover {
transform: translateY(-8px);
background: rgba(255, 255, 255, 0.25);
box-shadow: 0 10px 30px rgba(0,0,0,0.2);
}
.stat-number {
font-size: 48px;
font-weight: 800;
margin-bottom: 10px;
display: block;
}
.stat-label {
font-size: 16px;
opacity: 0.9;
font-weight: 500;
}
/* ===== PROBLEM/SOLUTION BOXES ===== */
.problem-showcase {
background: linear-gradient(to right, #fff3cd, #fff8e1);
padding: 35px;
border-radius: 16px;
margin: 35px 0;
border-left: 6px solid #ffc107;
box-shadow: 0 5px 20px rgba(255, 193, 7, 0.2);
}
.solution-showcase {
background: linear-gradient(to right, #d1ecf1, #e7f5f8);
padding: 35px;
border-radius: 16px;
margin: 35px 0;
border-left: 6px solid #17a2b8;
box-shadow: 0 5px 20px rgba(23, 162, 184, 0.2);
}
.scenario-box {
background: white;
padding: 25px;
border-radius: 12px;
margin: 20px 0;
box-shadow: 0 3px 15px rgba(0,0,0,0.1);
}
.recommendation-output {
font-size: 16px;
line-height: 1.6;
}
.metric-card {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 30px 20px;
border-radius: 16px;
text-align: center;
box-shadow: 0 8px 24px rgba(102, 126, 234, 0.3);
transition: transform 0.3s ease, box-shadow 0.3s ease;
margin: 10px;
}
.metric-card:hover {
transform: translateY(-5px);
box-shadow: 0 12px 32px rgba(102, 126, 234, 0.4);
}
.metric-card h2 {
font-size: 48px;
font-weight: 700;
margin: 0 0 10px 0;
color: white;
}
.metric-card p {
font-size: 16px;
margin: 0;
opacity: 0.9;
color: white;
}
.metric-card-green {
background: linear-gradient(135deg, #11998e 0%, #38ef7d 100%);
}
.metric-card-orange {
background: linear-gradient(135deg, #f093fb 0%, #f5576c 100%);
}
.metric-card-blue {
background: linear-gradient(135deg, #4facfe 0%, #00f2fe 100%);
}
/* ===== ANALYTICS TAB - WARNING BOX ===== */
.current-month-warning {
background: linear-gradient(135deg, #fff4e6 0%, #ffe8cc 100%);
border-left: 4px solid #ff9800;
padding: 15px 20px;
border-radius: 8px;
margin: 20px 0;
box-shadow: 0 2px 8px rgba(255, 152, 0, 0.2);
}
.current-month-warning h4 {
color: #e65100;
margin: 0 0 10px 0;
font-size: 18px;
font-weight: 600;
}
.current-month-warning p {
color: #5d4037;
margin: 5px 0;
font-size: 14px;
}
.thinking-dots {
display: inline-flex;
align-items: center;
gap: 8px;
padding: 20px;
font-size: 18px;
color: #667eea;
font-weight: 500;
}
.thinking-dots::after {
content: '●●●';
display: inline-block;
letter-spacing: 4px;
animation: thinking 1.4s infinite;
color: #667eea;
}
@keyframes thinking {
0%, 20% {
content: '●○○';
}
40% {
content: '●●○';
}
60%, 100% {
content: '●●●';
}
}
/* Alternative bouncing dots animation */
.thinking-bounce {
display: inline-flex;
align-items: center;
gap: 5px;
}
.thinking-bounce span {
width: 10px;
height: 10px;
background: #667eea;
border-radius: 50%;
animation: bounce 1.4s infinite ease-in-out;
}
.thinking-bounce span:nth-child(1) {
animation-delay: 0s;
}
.thinking-bounce span:nth-child(2) {
animation-delay: 0.2s;
}
.thinking-bounce span:nth-child(3) {
animation-delay: 0.4s;
}
@keyframes bounce {
0%, 80%, 100% {
transform: translateY(0);
opacity: 0.5;
}
40% {
transform: translateY(-10px);
opacity: 1;
}
}
/* Pulsing effect */
.thinking-pulse {
display: inline-block;
animation: pulse 1.5s ease-in-out infinite;
}
@keyframes pulse {
0%, 100% {
opacity: 1;
}
50% {
opacity: 0.3;
}
}
/* ===== FORECAST TAB - PREDICTION BOX ===== */
.forecast-prediction {
background: linear-gradient(135deg, #e3f2fd 0%, #bbdefb 100%);
border-left: 4px solid #2196f3;
padding: 20px;
border-radius: 8px;
margin: 20px 0;
box-shadow: 0 2px 8px rgba(33, 150, 243, 0.2);
}
.forecast-prediction h3 {
color: #0d47a1;
margin: 0 0 15px 0;
font-size: 22px;
font-weight: 700;
}
.forecast-prediction .confidence-badge {
display: inline-block;
padding: 4px 12px;
background: #4caf50;
color: white;
border-radius: 12px;
font-size: 12px;
font-weight: 600;
margin-left: 10px;
}
/* ===== SECTION DIVIDERS ===== */
.section-divider {
border: 0;
height: 2px;
background: linear-gradient(to right, transparent, #ddd, transparent);
margin: 30px 0;
}
/* ===== INFO BOXES ===== */
.info-box {
background: #f5f5f5;
border-radius: 8px;
padding: 15px;
margin: 15px 0;
border-left: 3px solid #667eea;
}
.info-box-icon {
font-size: 24px;
margin-right: 10px;
vertical-align: middle;
}
table {
width: 100%;
border-collapse: collapse;
margin: 20px 0;
background: white;
border-radius: 8px;
overflow: hidden;
box-shadow: 0 2px 8px rgba(0,0,0,0.1);
}
table th {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 12px;
text-align: left;
font-weight: 600;
}
table td {
padding: 12px;
border-bottom: 1px solid #f0f0f0;
}
table tr:last-child td {
border-bottom: none;
}
table tr:hover {
background: #f8f9fa;
}
""",
) as app:
# ==================== HERO SECTION ====================
gr.HTML("""
🎯 Stop Losing Money at Every Purchase
You have 5 credit cards. You're at checkout. Which one do you use?
Most people pick wrong and lose $400+ per year.
Our AI agent makes the optimal choice in 2 seconds.
2s
Decision Time
(vs. 5 min manual)
35%
More Rewards
Earned
$400+
Saved Per Year
Per User
100%
Optimal Choice
Every Time
""")
# ==================== PROBLEM STORYTELLING ====================
gr.HTML("""
😰 The $400/Year Problem Nobody Talks About
📍 Real Scenario: Sunday Grocery Shopping
You're at Whole Foods with $127.50 of groceries.
You pull out your wallet and see 5 credit cards...
⏰ You Have 10 Seconds to Decide...
❓ Which card gives best rewards?
❓ Have you hit spending caps this month?
❓ Is 4x points better than 5% cashback?
❓ People are waiting behind you...
❌ What Usually Happens:
You panic and use your "default" card
Money lost on this transaction: -$3.19
This happens 50+ times per month.
Annual loss: $400-600
""")
# ==================== SOLUTION DEMONSTRATION ====================
gr.HTML("""
✨ Our AI Solution: Your Personal Rewards Optimizer
🤖 Same Scenario, With AI Agent
✅ Result:
💳 Use: Amex Gold
🎁 Earn: $5.10 (vs. $1.91 with default card)
⚡ Decision time: 2 seconds (vs. 5 minutes)
💡 Confidence: 100%
💰 You just saved $3.19 in 2 seconds with ZERO mental effort
""")
# ==================== VALUE PROPOSITION ====================
gr.Markdown("""
## 🌟 What Makes This a Winning Solution?
| Traditional Approach | Our AI Solution | Impact |
|---------------------|-----------------|---------|
| 😰 Manual calculation (5 min) | ⚡ AI decision (2 sec) | **150x faster** |
| 🤔 Mental math & guessing | 🎯 100% optimal choice | **35% more rewards** |
| 📝 Manual cap tracking | 🤖 Automatic monitoring | **Zero effort** |
| ❌ No explanations | 💡 Clear reasoning | **Build trust** |
| 📊 Reactive only | 🔮 Predictive insights | **Proactive optimization** |
---
### 💎 **Unique Differentiators**
#### 1️⃣ **Real-Time Transaction Intelligence**
- Not just a card comparison tool
- **Context-aware recommendations** at point of purchase
- Considers YOUR specific spending patterns and caps
#### 2️⃣ **Multi-Agent AI Architecture**
- Orchestrator coordinates multiple specialized agents
- **Reasoning engine** explains every decision
- Learns and adapts to user behavior
#### 3️⃣ **Predictive Optimization**
- Forecasts next month spending by category
- **Warns before hitting caps**
- Suggests optimal card rotation strategies
#### 4️⃣ **Practical & Immediate Value**
- Solves a **real pain point** everyone faces
- **Measurable ROI**: $400+ saved per year
- Works with existing cards (no signup needed)
---
### 🚀 **Ready to Stop Losing Money?**
⬇️ **Try the "Get Recommendation" tab below** to experience the magic yourself ⬇️
""")
# Agent status (keep your existing one)
agent_status = """
🤖 **Autonomous Agent:** ✅ Active (Claude 3.5 Sonnet)
📊 **Mode:** Dynamic Planning + Reasoning
⚡ **Services:** Smart Wallet + RAG + Forecast
"""
gr.Markdown(agent_status)
with gr.Tabs():
# ==================== TAB 1: GET RECOMMENDATION ====================
with gr.Tab("🎯 Get Recommendation"):
# ADD THIS CONTEXT BANNER
gr.HTML("""
💡 Experience the Magic: Real-Time AI Optimization
Simulate a real transaction: You're about to make a purchase.
Instead of spending 5 minutes calculating or guessing, let our AI agent analyze
your entire wallet and recommend the optimal card in under 2 seconds.
🎯 Try these scenarios:
- 🛒 Whole Foods, Groceries, $127.50
- 🍕 DoorDash, Restaurants, $45.00
- ⛽ Shell, Gas, $60.00
- ✈️ United Airlines, Travel, $450.00
""")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### Transaction Details")
user_dropdown = gr.Dropdown(
choices=SAMPLE_USERS,
value=SAMPLE_USERS[0],
label="User ID",
info="Select a user"
)
category_dropdown = gr.Dropdown(
choices=list(MCC_CATEGORIES.keys()),
value="Groceries",
label="🏷️ Type of Purchase",
info="Select the category first"
)
merchant_dropdown = gr.Dropdown(
choices=MERCHANTS_BY_CATEGORY["Groceries"],
value="Whole Foods",
label="🏪 Merchant Name",
info="Select merchant (changes based on category)",
allow_custom_value=True
)
amount_input = gr.Number(
label="💵 Amount (USD)",
value=125.50,
minimum=0.01,
step=0.01
)
with gr.Accordion("🤖 AI Settings", open=False):
use_gemini = gr.Checkbox(
label="Use Google Gemini for explanations",
value=False,
info="Switch to Gemini 1.5 Pro for AI insights"
)
date_input = gr.Textbox(
label="📅 Transaction Date (Optional)",
placeholder="YYYY-MM-DD or leave blank for today",
value=""
)
with gr.Accordion("⚙️ Advanced Options", open=False):
use_custom_mcc = gr.Checkbox(
label="Use Custom MCC Code",
value=False
)
custom_mcc_input = gr.Textbox(
label="Custom MCC Code",
placeholder="e.g., 5411",
visible=False,
interactive=True
)
def toggle_custom_mcc(use_custom):
return gr.update(visible=use_custom, interactive=use_custom)
use_custom_mcc.change(
fn=toggle_custom_mcc,
inputs=[use_custom_mcc],
outputs=[custom_mcc_input]
)
recommend_btn = gr.Button(
"🚀 Get Recommendation",
variant="primary",
size="lg"
)
with gr.Column(scale=2):
gr.Markdown("### 💡 Recommendation")
recommendation_output = gr.Markdown(
value="✨ Select a category and merchant, then click 'Get Recommendation'",
elem_classes=["recommendation-output"]
)
recommendation_chart = gr.Plot()
def update_merchant_choices(category):
"""Update merchant dropdown based on selected category"""
merchants = MERCHANTS_BY_CATEGORY.get(category, ["Custom Merchant"])
return gr.update(
choices=merchants,
value=merchants[0] if merchants else ""
)
category_dropdown.change(
fn=update_merchant_choices,
inputs=[category_dropdown],
outputs=[merchant_dropdown]
)
with gr.Row():
with gr.Column():
gr.Markdown("### 📊 Quick Stats")
stats_output = gr.Markdown()
with gr.Column():
gr.Markdown("### 🔄 Card Comparison")
comparison_output = gr.Markdown()
recommend_btn.click(
fn=get_recommendation_with_agent,
inputs=[user_dropdown, merchant_dropdown, category_dropdown, amount_input],
outputs=[recommendation_output, recommendation_chart]
)
gr.Markdown("### 📝 Example Transactions")
gr.Examples(
examples=EXAMPLES,
inputs=[
user_dropdown,
category_dropdown,
merchant_dropdown,
amount_input,
use_custom_mcc,
custom_mcc_input,
date_input
],
outputs=[
recommendation_output,
comparison_output,
stats_output
],
fn=get_recommendation,
cache_examples=False
)
# ==================== TAB 2: SMART WALLET ====================
with gr.Tab("💳 Smart Wallet"):
gr.Markdown("## Your Credit Card Portfolio")
wallet_user = gr.Dropdown(
choices=SAMPLE_USERS,
value=SAMPLE_USERS[0],
label="👤 Select User"
)
refresh_wallet_btn = gr.Button("🔄 Refresh Wallet", variant="secondary")
wallet_output = gr.Markdown(value="*Loading wallet...*")
wallet_chart = gr.Plot()
def update_wallet(user_id):
return load_user_wallet(user_id)
wallet_user.change(
fn=update_wallet,
inputs=[wallet_user],
outputs=[wallet_output, wallet_chart]
)
refresh_wallet_btn.click(
fn=update_wallet,
inputs=[wallet_user],
outputs=[wallet_output, wallet_chart]
)
app.load(
fn=update_wallet,
inputs=[wallet_user],
outputs=[wallet_output, wallet_chart]
)
# ==================== TAB 3: ANALYTICS ====================
with gr.Tab("📊 Analytics"):
gr.Markdown("## 🎯 Your Rewards Optimization Dashboard")
with gr.Row():
analytics_user = gr.Dropdown(
choices=SAMPLE_USERS,
value=SAMPLE_USERS[0],
label="👤 View Analytics For User",
scale=3
)
refresh_analytics_btn = gr.Button(
"🔄 Refresh Analytics",
variant="secondary",
scale=1
)
metrics_display = gr.HTML(
value="""
$0
💰 Potential Annual Savings
0%
📈 Rewards Rate Increase
0
✅ Optimized Transactions
0/100
⭐ Optimization Score
"""
)
gr.Markdown("---")
gr.Markdown("## 📊 Visual Analytics")
with gr.Row():
with gr.Column(scale=2):
spending_chart = gr.Plot(label="Spending vs Rewards")
with gr.Column(scale=1):
optimization_gauge = gr.Plot(label="Your Score")
with gr.Row():
with gr.Column(scale=1):
rewards_pie_chart = gr.Plot(label="Rewards Distribution")
with gr.Column(scale=1):
card_performance_chart = gr.Plot(label="Top Performing Cards")
with gr.Row():
trend_chart = gr.Plot(label="12-Month Trends")
gr.Markdown("---")
gr.Markdown("## 📋 Detailed Breakdown")
with gr.Row():
with gr.Column(scale=1):
gr.Markdown("### 💰 Category Spending Breakdown")
spending_table = gr.Markdown(
value="*Loading data...*"
)
with gr.Column(scale=1):
gr.Markdown("### 📈 Monthly Trends & Insights")
insights_display = gr.Markdown(
value="*Loading insights...*"
)
# ===== CHANGED SECTION: Current Month Summary =====
gr.HTML('
')
gr.Markdown("""
📊
Current Month Summary - Quick insights based on your spending so far this month
""")
current_month_summary = gr.HTML(
value="""
⚠️ This Month's Insights
Loading current month data...
""",
label=None
)
# Add clear call-to-action to Forecast tab
gr.Markdown("""
Want to see next month's predictions and optimization strategies?
👉
Go to the Forecast Tab above →
""")
analytics_status = gr.Markdown(
value="*Select a user to view analytics*",
elem_classes=["status-text"]
)
# Event handlers
analytics_user.change(
fn=update_analytics_with_charts,
inputs=[analytics_user],
outputs=[
metrics_display,
spending_chart,
optimization_gauge,
rewards_pie_chart,
card_performance_chart,
trend_chart,
spending_table,
insights_display,
current_month_summary, # Changed from forecast_display
analytics_status
]
)
refresh_analytics_btn.click(
fn=update_analytics_with_charts,
inputs=[analytics_user],
outputs=[
metrics_display,
spending_chart,
optimization_gauge,
rewards_pie_chart,
card_performance_chart,
trend_chart,
spending_table,
insights_display,
current_month_summary, # Changed from forecast_display
analytics_status
]
)
app.load(
fn=update_analytics_with_charts,
inputs=[analytics_user],
outputs=[
metrics_display,
spending_chart,
optimization_gauge,
rewards_pie_chart,
card_performance_chart,
trend_chart,
spending_table,
insights_display,
current_month_summary, # Changed from forecast_display
analytics_status
]
)
# ==================== TAB 4: FORECAST ====================
with gr.Tab("📈 Forecast"):
# Add clear header with explanation
gr.Markdown("""
🔮 AI-Powered Spending Forecast
Machine learning predictions for your next 1-3 months
with personalized optimization strategies
""")
gr.Markdown("""
🤖
How it works: Our AI analyzes your historical spending patterns,
seasonal trends, and card benefits to predict future spending and recommend
the best cards to maximize your rewards.
""")
with gr.Row():
forecast_user = gr.Dropdown(
choices=SAMPLE_USERS,
value=SAMPLE_USERS[0],
label="👤 Select User"
)
refresh_forecast_btn = gr.Button(
"🔄 Refresh Forecast",
variant="primary",
size="sm"
)
# CHANGED: gr.HTML -> gr.Markdown
forecast_output = gr.Markdown(value="*Loading forecast...*")
forecast_chart = gr.Plot()
def update_forecast(user_id):
return load_user_forecast(user_id)
forecast_user.change(
fn=update_forecast,
inputs=[forecast_user],
outputs=[forecast_output, forecast_chart]
)
refresh_forecast_btn.click(
fn=update_forecast,
inputs=[forecast_user],
outputs=[forecast_output, forecast_chart]
)
app.load(
fn=update_forecast,
inputs=[forecast_user],
outputs=[forecast_output, forecast_chart]
)
# ==================== TAB 5: ASK AI ====================
with gr.Tab("💬 Ask AI"):
gr.Markdown("## Chat with RewardPilot AI")
gr.Markdown("*Ask questions about credit cards, rewards, and your spending*")
chatbot = gr.Chatbot(height=400, label="AI Assistant")
with gr.Row():
msg = gr.Textbox(
placeholder="Ask me anything about credit cards...",
label="Your Question",
scale=4
)
send_btn = gr.Button("Send", variant="primary", scale=1)
chat_user = gr.Dropdown(
choices=["u_alice", "u_bob", "u_charlie"],
label="Your Profile",
value="u_alice",
visible=True
)
def respond(message, chat_history, user_id):
"""Handle chat responses with error handling"""
if not message.strip():
return "", chat_history
user_context = {}
try:
analytics = client.get_user_analytics(user_id)
if analytics.get('success'):
data = analytics.get('data', {})
user_context = {
'cards': safe_get(data, 'cards', ['Amex Gold', 'Chase Sapphire Reserve']),
'monthly_spending': safe_get(data, 'total_spending', 0),
'top_category': safe_get(data, 'top_category', 'Groceries')
}
except Exception as e:
print(f"Error getting user context: {e}")
user_context = {
'cards': ['Amex Gold', 'Chase Sapphire Reserve'],
'monthly_spending': 3450.75,
'top_category': 'Groceries'
}
try:
if config.LLM_ENABLED:
bot_response = llm.chat_response(message, user_context, chat_history)
else:
bot_response = "I'm currently in fallback mode. Ask me about specific cards or categories!"
except Exception as e:
print(f"Chat error: {e}")
bot_response = f"I encountered an error. Please try asking your question differently."
chat_history.append((message, bot_response))
return "", chat_history
msg.submit(respond, [msg, chatbot, chat_user], [msg, chatbot])
send_btn.click(respond, [msg, chatbot, chat_user], [msg, chatbot])
gr.Markdown("### 💡 Try asking:")
gr.Examples(
examples=[
["Which card should I use at Costco?"],
["How can I maximize my grocery rewards?"],
["What's the best travel card for international trips?"],
["Tell me about the Amex Gold card"],
["Am I close to any spending caps?"],
],
inputs=[msg]
)
# ==================== TAB 6: RESOURCES (About + Agent Insight + API Docs) ====================
with gr.Tab("ℹ️ Resources"):
with gr.Tabs():
# ========== SUB-TAB: ABOUT ==========
with gr.Tab("📖 About"):
gr.Markdown(
"""
## 🎯 About RewardPilot
### 🚀 The Vision
**Stop leaving money on the table.** Most people use the same 1-2 credit cards for
everything, missing out on hundreds of dollars in rewards annually. But manually
calculating the optimal card for every purchase is impractical.
**That's where AI comes in.**
---
### 💡 The Problem We Solve
#### Real-World Scenario:
You're standing at a checkout counter with **5 credit cards** in your wallet.
**The Question:** Which card should you use for this $85 grocery purchase?
#### Manual Calculation (What Most People Do):
1. Remember reward rates for all 5 cards ❌ (takes 30+ seconds)
2. Check if you've hit spending caps this month ❌ (requires tracking)
3. Calculate actual rewards for each card ❌ (mental math)
4. Consider special promotions or bonuses ❌ (easy to forget)
5. Make a decision before people behind you get annoyed ❌ (pressure!)
**Result:** You pick your "default" card and lose $3.40 in rewards on this single transaction.
**Annual Impact:** Losing $15-50/month = **$180-600/year** in missed rewards.
---
### ✨ Our AI-Powered Solution
#### How It Works:
```
📱 INPUT (takes 10 seconds)
├─ Merchant: "Whole Foods"
├─ Category: "Groceries"
└─ Amount: "$85.00"
🤖 AI AGENT ANALYZES (takes 2 seconds)
├─ Your 5 credit cards and reward structures
├─ Current spending: $450/$1500 on Amex Gold groceries
├─ Citi Custom Cash already hit $500 cap this month
├─ No active promotional bonuses
└─ Historical pattern: You shop at Whole Foods 2x/week
✅ RECOMMENDATION (instant)
├─ 💳 Use: Amex Gold
├─ 🎁 Earn: $3.40 (4x points = 340 points)
├─ 💡 Reason: "Best grocery multiplier, you haven't hit annual cap"
└─ ⚠️ Warning: "You'll hit $1500 monthly cap in 3 more transactions"
```
#### The Result:
- ⚡ **Decision time:** 2 seconds (vs. 2-5 minutes manually)
- 💰 **Rewards:** $3.40 earned (vs. $1.28 with default card)
- 🎯 **Accuracy:** 100% optimal choice every time
- 🧠 **Mental effort:** Zero (AI does all the thinking)
---
### 📊 Real-World Impact
#### Case Study: Sample User "Alice"
**Before Using Our System:**
- Used Chase Freedom Unlimited for everything (1.5% cashback)
- Annual rewards: **$450**
- Hit quarterly caps early and didn't realize
- Missed travel bonuses on Sapphire Reserve
**After Using Our System (3 months):**
- Uses optimal card for each transaction
- Projected annual rewards: **$680**
- AI warned about caps and suggested card rotation
- Activated travel bonuses at right time
**Result:** **+$230/year (51% increase)** with zero extra effort
---
### 🏗️ Architecture
- 🎯 **Model Context Protocol (MCP)** architecture
- 🤖 **LLM-powered explanations** using Claude 3.5 Sonnet
- 📚 **RAG (Retrieval-Augmented Generation)** for card benefits
- 📈 **ML-based spending forecasts**
- 📊 **Interactive visualizations**
---
### 🔧 Technology Stack
- **Backend:** FastAPI, Python
- **Frontend:** Gradio
- **AI/ML:** Multi-agent system with RAG
- **LLM:** Claude 3.5 Sonnet (Anthropic)
- **Architecture:** MCP (Model Context Protocol)
- **Deployment:** Hugging Face Spaces
---
### 🎓 Built For
**MCP 1st Birthday Hackathon** - Celebrating one year of the Model Context Protocol
---
**Ready to maximize your rewards?** Start with the "Get Recommendation" tab! 🚀
"""
)
# ========== SUB-TAB: AGENT INSIGHT ==========
with gr.Tab("🔍 Agent Insight"):
gr.Markdown("""
## How the Autonomous Agent Works
RewardPilot uses **Claude 3.5 Sonnet** as an autonomous agent to provide intelligent card recommendations.
### 🎯 **Phase 1: Planning**
The agent analyzes your transaction and decides:
- Which microservices to call (Smart Wallet, RAG, Forecast)
- In what order to call them
- What to optimize for (rewards, caps, benefits)
- Confidence level of the plan
### 🤔 **Phase 2: Execution**
The agent dynamically:
- Calls services based on the plan
- Handles failures gracefully
- Adapts if services are unavailable
- Collects all relevant data
### 🧠 **Phase 3: Reasoning**
The agent synthesizes results to:
- Explain **why** this card is best
- Identify potential risks or warnings
- Suggest alternative options
- Calculate annual impact
### 📚 **Phase 4: Learning**
The agent improves over time by:
- Storing past decisions
- Learning from user feedback
- Adjusting strategies for similar transactions
- Building a knowledge base
---
### 🔑 **Key Features**
✅ **Natural Language Explanations** - Understands context like a human
✅ **Dynamic Planning** - Adapts to your specific situation
✅ **Confidence Scoring** - Tells you how certain it is
✅ **Multi-Service Coordination** - Orchestrates 3 microservices
✅ **Self-Correction** - Learns from mistakes
---
### 📊 **Example Agent Plan**
```json
{
"strategy": "Optimize for grocery rewards with cap monitoring",
"service_calls": [
{"service": "smart_wallet", "priority": 1, "reason": "Get base recommendation"},
{"service": "spend_forecast", "priority": 2, "reason": "Check spending caps"},
{"service": "rewards_rag", "priority": 3, "reason": "Get detailed benefits"}
],
"confidence": 0.92,
"expected_outcome": "Recommend Amex Gold for 4x grocery points"
}
```
---
### 🎓 **Powered By**
- **Model**: Claude 3.5 Sonnet (Anthropic)
- **Architecture**: Autonomous Agent Pattern
- **Framework**: LangChain + Custom Logic
- **Memory**: Redis (for learning)
---
**Try it out in the "Get Recommendation" tab!** 🚀
""")
# ========== SUB-TAB: API DOCS ==========
with gr.Tab("📚 API Documentation"):
api_docs_html = """
📡 API Endpoints
Orchestrator API
Base URL: https://mcp-1st-birthday-rewardpilot-orchestrator.hf.space
POST /recommend
Get comprehensive card recommendation.
{
"user_id": "u_alice",
"merchant": "Whole Foods",
"mcc": "5411",
"amount_usd": 125.50,
"transaction_date": "2025-01-15"
}
GET /analytics/{user_id}
Get user analytics and spending insights.
GET /analytics/u_alice
Other Services
- Smart Wallet: https://mcp-1st-birthday-rewardpilot-smart-wallet.hf.space
- Rewards-RAG: https://mcp-1st-birthday-rewardpilot-rewards-rag.hf.space
- Spend-Forecast: https://mcp-1st-birthday-rewardpilot-spend-forecast.hf.space
📚 Interactive Docs
Visit /docs on any service for Swagger UI:
🔧 cURL Example
curl -X POST https://mcp-1st-birthday-rewardpilot-orchestrator.hf.space/recommend \\
-H "Content-Type: application/json" \\
-d '{
"user_id": "u_alice",
"merchant": "Whole Foods",
"mcc": "5411",
"amount_usd": 125.50
}'
🐍 Python Example
import requests
url = "https://mcp-1st-birthday-rewardpilot-orchestrator.hf.space/recommend"
payload = {
"user_id": "u_alice",
"merchant": "Whole Foods",
"mcc": "5411",
"amount_usd": 125.50
}
response = requests.post(url, json=payload)
print(response.json())
📋 Response Format
{
"recommended_card": "c_amex_gold",
"rewards_earned": 5.02,
"rewards_rate": "4x points",
"confidence": 0.95,
"reasoning": "Amex Gold offers 4x points on groceries...",
"alternative_options": [
{
"card": "c_citi_custom_cash",
"reward_amount": 6.28,
"reason": "5% cashback on groceries..."
}
],
"warnings": [
"You're approaching your $500 monthly cap"
]
}
🔐 Authentication
Currently, the API is open for demo purposes. In production, you would need:
- API Key in headers:
X-API-Key: your_key_here
- OAuth 2.0 for user-specific data
⚡ Rate Limits
- Free Tier: 100 requests/hour
- Pro Tier: 1000 requests/hour
- Enterprise: Unlimited
❓ Support
For API support, please visit our GitHub repository or contact support.
"""
gr.HTML(api_docs_html)
# ========== SUB-TAB: FAQs ==========
with gr.Tab("❓ FAQs"):
gr.Markdown("""
## Frequently Asked Questions
### General Questions
**Q: What is RewardPilot?**
A: RewardPilot is an AI-powered system that recommends the best credit card to use for each transaction to maximize your rewards.
**Q: How does it work?**
A: It analyzes your transaction details (merchant, amount, category) against your credit card portfolio and recommends the card that will earn you the most rewards.
**Q: Is my data secure?**
A: Yes! All data is encrypted and we follow industry-standard security practices. We never store sensitive card information like CVV or full card numbers.
---
### Using the System
**Q: How accurate are the recommendations?**
A: Our AI agent has a 95%+ confidence rate for most recommendations. The system considers reward rates, spending caps, and category bonuses.
**Q: What if I don't have the recommended card?**
A: The system shows alternative options from your wallet. You can also view the "Alternative Options" section for other good choices.
**Q: Can I add custom MCC codes?**
A: Yes! Use the "Advanced Options" section in the Get Recommendation tab to enter custom MCC codes.
---
### Analytics & Forecasts
**Q: How is the optimization score calculated?**
A: It's based on reward rates (30%), cap availability (25%), annual fee value (20%), category match (20%), and penalties (5%).
**Q: How accurate are the spending forecasts?**
A: Our ML models achieve 85-92% accuracy based on your historical spending patterns.
**Q: Can I export my analytics data?**
A: This feature is coming soon! You'll be able to export to CSV and PDF.
---
### Technical Questions
**Q: What APIs does RewardPilot use?**
A: We use 4 main services: Orchestrator, Smart Wallet, Rewards-RAG, and Spend-Forecast.
**Q: Can I integrate RewardPilot into my app?**
A: Yes! Check the API Documentation tab for integration details.
**Q: What LLM powers the AI agent?**
A: We use Claude 3.5 Sonnet by Anthropic for intelligent reasoning and explanations.
---
### Troubleshooting
**Q: Why am I seeing "Demo Mode" warnings?**
A: This means the system is using mock data. Ensure the orchestrator API is connected.
**Q: The recommendation seems wrong. Why?**
A: Check the "Agent Insight" tab to see the reasoning. If you still think it's wrong, please report it.
**Q: How do I report a bug?**
A: Please open an issue on our [GitHub repository](https://github.com/your-repo).
---
**Still have questions?** Contact us at support@rewardpilot.ai
""")
# ===================== Launch App =====================
if __name__ == "__main__":
app.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
)