""" main.py ------- Phase 3 Section 5 — FastAPI dashboard for RingGuard. Endpoints: GET / — HTML dashboard: table of all cases GET /cases — JSON list of all cases (from cases_demo.json) GET /cases/{case_id} — JSON detail for one case + full audit chain GET /health — health check The dashboard reads directly from cases_demo.json and audit_log.jsonl. No database, no auth — demo only. Usage: uvicorn api.main:app --reload --port 7860 """ import json import os from typing import Optional from fastapi import FastAPI, HTTPException from fastapi.responses import HTMLResponse, JSONResponse BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) CASES_PATH = os.path.join(BASE_DIR, "data", "synthetic", "cases_demo.json") AUDIT_PATH = os.path.join(BASE_DIR, "audit", "audit_log.jsonl") app = FastAPI( title="RingGuard Dashboard", description="Phase 3 — Fraud ring detection results and audit trail", version="3.0.0", ) # --------------------------------------------------------------------------- # Data loaders # --------------------------------------------------------------------------- def _load_cases() -> list: if not os.path.exists(CASES_PATH): return [] with open(CASES_PATH, encoding="utf-8") as f: return json.load(f) def _load_audit() -> list: if not os.path.exists(AUDIT_PATH): return [] records = [] with open(AUDIT_PATH, encoding="utf-8") as f: for line in f: line = line.strip() if line: records.append(json.loads(line)) return records # --------------------------------------------------------------------------- # API endpoints # --------------------------------------------------------------------------- @app.get("/health") def health(): return {"status": "ok", "version": "3.0.0"} @app.get("/cases", response_class=JSONResponse) def get_cases(): """Return all cases as a JSON list.""" cases = _load_cases() audit_all = _load_audit() # Build a lookup: case_id -> first audit record audit_map = {} for r in audit_all: cid = r.get("case_id") if cid and cid not in audit_map: audit_map[cid] = r # Return a summary view (without full narrative for list) summary = [] for c in cases: case_id = c.get("case_id") # Prefer action_taken from audit record when case has null action = c.get("action_taken") or audit_map.get(case_id, {}).get("action_taken") or {} summary.append({ "case_id": case_id, "cluster_id": c.get("cluster_id"), "cluster_size": len(c.get("customer_ids", [])), "risk_score": round(c.get("risk_probability", 0), 4), "policy_decision": c.get("policy_decision"), "action_taken": bool(action.get("order_id")), "timestamp": c.get("timestamp"), }) return summary @app.get("/cases/{case_id}", response_class=JSONResponse) def get_case_detail(case_id: str): """Return full detail for one case, including audit chain.""" cases = _load_cases() case = next((c for c in cases if c.get("case_id") == case_id), None) if not case: raise HTTPException(status_code=404, detail=f"Case {case_id} not found") # Find audit records for this case audit_records = [r for r in _load_audit() if r.get("case_id") == case_id] return { "case": case, "audit_chain": audit_records, } # --------------------------------------------------------------------------- # HTML dashboard # --------------------------------------------------------------------------- def _decision_badge(decision: str) -> str: colours = { "auto_hold": "#c0392b", "escalate": "#e67e22", "log_only": "#27ae60", } colour = colours.get(decision, "#7f8c8d") return ( f'' f'{decision.upper()}' ) def _build_dashboard(cases: list, audit_map: dict = None) -> str: if audit_map is None: audit_map = {} rows = "" for c in cases: decision = c.get("policy_decision", "") badge = _decision_badge(decision) risk = round(c.get("risk_probability", 0) * 100, 1) cluster = c.get("cluster_id", "") case_id = c.get("case_id", "") size = len(c.get("customer_ids", [])) # Prefer action_taken from audit record when case JSON has null action = c.get("action_taken") or audit_map.get(case_id, {}).get("action_taken") or {} order_id = action.get("order_id") or "—" verified_val = action.get("notes_verified") if verified_val is True: verified = '✓ Yes' elif action: verified = 'No' else: verified = "—" rows += ( f'' f'{case_id}' f'{cluster}' f'{size}' f'{risk}%' f'{badge}' f'{order_id}' f'{verified}' f'\n' ) auto_holds = sum(1 for c in cases if c.get("policy_decision") == "auto_hold") escalates = sum(1 for c in cases if c.get("policy_decision") == "escalate") log_only = sum(1 for c in cases if c.get("policy_decision") == "log_only") return f""" RingGuard — Fraud Detection Dashboard

RingGuard — Fraud Detection Dashboard

Phase 3 · {len(cases)} cases · LightGBM + LangGraph + Razorpay test-mode

Auto Hold
{auto_holds}
Escalate
{escalates}
Log Only
{log_only}
Total Cases
{len(cases)}
{rows}
Case Cluster Size Risk Score Decision Razorpay Order Verified

Click any row to view the full evidence bundle, LLM narrative, and audit chain.

Built by Rohit · RingGuard v3.0
""" @app.get("/", response_class=HTMLResponse) def dashboard(): """Serve the HTML dashboard.""" cases = _load_cases() audit_all = _load_audit() audit_map = {} for r in audit_all: cid = r.get("case_id") if cid and cid not in audit_map: audit_map[cid] = r html = _build_dashboard(cases, audit_map) return HTMLResponse(content=html) def _build_cluster_svg(customer_ids: list, strong_link_count: int, weak_link_count: int, formation_type: str, decision: str) -> str: """ Build a simple SVG diagram of the cluster: - One circle per customer, arranged in a ring layout - Red lines for strong links (shared device/instrument), grey dashed for weak (address) - Node colour reflects decision """ import math n = len(customer_ids) if n == 0: return "" W, H = 420, 260 cx, cy = W // 2, H // 2 r = min(cx, cy) - 44 node_colour = { "auto_hold": "#c0392b", "escalate": "#e67e22", "log_only": "#27ae60", }.get(decision, "#7c5cd8") # Place nodes on a circle; single node goes centre angles = [2 * math.pi * i / n - math.pi / 2 for i in range(n)] positions = [(cx + r * math.cos(a), cy + r * math.sin(a)) for a in angles] if n > 1 else [(cx, cy)] lines = [] # Strong edges between every pair (fully connected if strong_link_count == n*(n-1)/2) # For simplicity: draw strong edges between consecutive nodes to represent links # Real topology is cluster-wide; we show it proportionally total_possible = n * (n - 1) // 2 if n > 1 else 0 drawn_strong = 0 drawn_weak = 0 if total_possible > 0: for i in range(n): for j in range(i + 1, n): x1, y1 = positions[i] x2, y2 = positions[j] if drawn_strong < strong_link_count: lines.append( f'' ) drawn_strong += 1 elif drawn_weak < weak_link_count: lines.append( f'' ) drawn_weak += 1 nodes = [] for i, (px, py) in enumerate(positions): label = customer_ids[i].replace("CUST", "C") nodes.append( f'' f'' f'{label}' ) legend = ( f'' f'' f'Strong link (device/instrument)' f'' f'' f'Weak link (address)' ) formation_label = formation_type.replace("_", " ").title() if formation_type else "" svg = ( f'' f'{"".join(lines)}' f'{"".join(nodes)}' f'{formation_label} · ' f'{strong_link_count} strong · {weak_link_count} weak' f'{legend}' f'' ) return svg @app.get("/cases/{case_id}/detail", response_class=HTMLResponse) def case_detail_html(case_id: str): """Serve HTML detail page for a single case.""" cases = _load_cases() case = next((c for c in cases if c.get("case_id") == case_id), None) if not case: raise HTTPException(status_code=404, detail=f"Case {case_id} not found") audit_records = [r for r in _load_audit() if r.get("case_id") == case_id] decision = case.get("policy_decision", "") badge = _decision_badge(decision) risk = round(case.get("risk_probability", 0) * 100, 2) # Prefer audit narrative/confidence when case JSON has null audit_rec0 = audit_records[0] if audit_records else {} narrative = (case.get("llm_narrative") or audit_rec0.get("llm_narrative") or "No narrative available for this case.") confidence = (case.get("llm_confidence_statement") or audit_rec0.get("llm_confidence_statement") or "—") members = ", ".join(case.get("customer_ids", [])) # Prefer action_taken from audit record when case JSON has null action = case.get("action_taken") or audit_rec0.get("action_taken") or {} order_id = action.get("order_id") or "—" verified = "Yes" if action.get("notes_verified") else "No" tool_used = action.get("tool") or "—" timestamp = action.get("timestamp") or "—" # Pull graph score inputs from audit record for the cluster diagram audit_rec = audit_rec0 score_inputs = audit_rec.get("graph_score_inputs", {}) strong_links = int(score_inputs.get("strong_link_count", case.get("strong_link_count", 0))) weak_links = int(score_inputs.get("weak_link_count", case.get("weak_link_count", 0))) # Resolve formation type label from strong_link_count formation_label = "strong_anchored" if strong_links > 0 else "weak_only" cluster_svg = _build_cluster_svg( case.get("customer_ids", []), strong_links, weak_links, formation_label, decision ) # Feature table rows from audit feature_keys = [ ("order_count", "Order Count"), ("total_amount", "Total Amount (Rs.)"), ("avg_order_amount", "Avg Order Amount (Rs.)"), ("return_rate", "Return Rate"), ("chargeback_rate", "Chargeback Rate"), ("shared_coupon_count", "Shared Coupons"), ("order_velocity", "Order Velocity (orders/day)"), ("time_span_days", "Time Span (days)"), ] feature_rows_html = "" if score_inputs: for key, label in feature_keys: val = score_inputs.get(key) if val is not None: if isinstance(val, float): display = f"{val:.3f}" if val < 100 else f"{val:,.0f}" else: display = str(val) feature_rows_html += f'{label}{display}' audit_rows_html = "" for rec in audit_records: ts = rec.get("timestamp", "")[:19].replace("T", " ") audit_rows_html += ( f'' f'{ts}' f'{rec.get("policy_decision","")}' f'{rec.get("reason_if_not_auto_actioned") or "—"}' f'\n' ) html = f""" RingGuard — {case_id}
← Back to all cases

{case_id} — {case.get("cluster_id","")}

Summary

Risk Score
{risk}%
Decision
{badge}
Cluster Size
{len(case.get("customer_ids",[]))} accounts
Formation
{formation_label.replace("_"," ")}
Members: {members}

Cluster Graph

{cluster_svg}

Evidence — Key Features

{f'{feature_rows_html}
' if feature_rows_html else '

No feature data available.

'}

LLM Investigator Narrative

{narrative}
Confidence: {confidence}

Razorpay Action (Test Mode)

Order ID
{order_id}
Notes Verified
{verified}
Tool Used
{tool_used}
Timestamp
{timestamp[:19] if len(timestamp)>=19 else timestamp}

Audit Chain

{f'{audit_rows_html}
TimestampDecisionReason
' if audit_rows_html else '

No audit records found for this case.

'}
Built by Rohit · RingGuard v3.0
""" return HTMLResponse(content=html)