server.py

"""FastAPI server exposing the Payment Credit Environment as an HTTP API."""

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Dict, Optional, Literal
from datetime import date, datetime, timedelta
import random
import uuid

app = FastAPI(
    title="payment_credit_env",
    version="0.2.0",
    description="OpenEnv-compatible payment credit decision environment for hackathon."
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

ActionType = Literal["approve_card_1", "approve_card_2", "route_to_debit", "deny_transaction", "request_more_info", "adjust_credit_limit", "offer_installments", "escalate_review"]
ALL_ACTIONS = ["approve_card_1", "approve_card_2", "route_to_debit", "deny_transaction", "request_more_info", "adjust_credit_limit", "offer_installments", "escalate_review"]

class TransactionState(BaseModel):
    transaction_id: str
    amount: float
    credit_score: int
    available_credit: float
    monthly_spend: float
    debt_to_income: float
    payment_history: float
    credit_utilization: float
    last_payment_date: str
    account_age_months: int


class StepRequest(BaseModel):
    action: ActionType


class LeaderboardEntry(BaseModel):
    action: str
    reward: float


class PolicyCheck(BaseModel):
    name: str
    passed: bool
    detail: str


class StepResponse(BaseModel):
    transaction_id: str
    action: str
    reward: float
    done: bool = False
    state: TransactionState
    risk_band: str
    recommended_action: str
    reasons: List[str]
    leaderboard: List[LeaderboardEntry]
    policy_checks: List[PolicyCheck]


class AuditLogEntry(BaseModel):
    timestamp: str
    transaction_id: str
    action: str
    reward: float
    risk_band: str
    recommended_action: str


class EnvStore:
    def __init__(self):
        self.current_state = None
        self.audit_log = []

    def reset(self):
        self.current_state = self._sample_transaction()
        self.audit_log = []
        return self.current_state

    def get_state(self):
        if self.current_state is None:
            self.current_state = self._sample_transaction()
        return self.current_state

    def _sample_transaction(self):
        tid = f"TXN{random.randint(1000, 9999)}"
        amt = round(random.uniform(500, 15000), 2)
        cs = random.randint(580, 780)
        avail = round(random.uniform(2000, 20000), 2)
        spend = round(random.uniform(300, 3000), 2)
        dti = round(random.uniform(0.15, 0.65), 2)
        ph = round(random.uniform(0.75, 1.0), 2)
        cu = round(random.uniform(0.2, 0.95), 2)
        last_pay = (date.today() - timedelta(days=random.randint(0, 60))).isoformat()
        age = random.randint(6, 120)
        return TransactionState(
            transaction_id=tid, amount=amt, credit_score=cs,
            available_credit=avail, monthly_spend=spend, debt_to_income=dti,
            payment_history=ph, credit_utilization=cu, last_payment_date=last_pay,
            account_age_months=age
        )


def get_risk_band(s: TransactionState) -> str:
    score = 0
    if s.credit_score >= 750: score += 2
    elif s.credit_score >= 680: score += 1
    if s.debt_to_income < 0.35: score += 1
    elif s.debt_to_income > 0.5: score -= 1
    if s.payment_history >= 0.95: score += 1
    elif s.payment_history < 0.85: score -= 1
    if s.credit_utilization < 0.3: score += 1
    elif s.credit_utilization > 0.7: score -= 1
    if s.account_age_months >= 36: score += 1
    if score >= 3: return "low"
    elif score >= 1: return "medium"
    else: return "high"


def recommended_action_for_band(s: TransactionState, risk_band: str) -> str:
    if risk_band == "low": return "approve_card_1"
    elif risk_band == "medium": return "offer_installments"
    else: return "escalate_review"


def build_reasons(s: TransactionState, risk_band: str) -> List[str]:
    reasons = []
    if s.credit_score >= 750: reasons.append(f"Excellent credit score ({s.credit_score})")
    elif s.credit_score < 650: reasons.append(f"Low credit score ({s.credit_score})")
    if s.debt_to_income < 0.35: reasons.append(f"Healthy debt-to-income ({s.debt_to_income:.1%})")
    elif s.debt_to_income > 0.5: reasons.append(f"High debt-to-income ({s.debt_to_income:.1%})")
    if s.payment_history >= 0.95: reasons.append(f"Strong payment history ({s.payment_history:.1%})")
    elif s.payment_history < 0.85: reasons.append(f"Weak payment history ({s.payment_history:.1%})")
    if s.credit_utilization < 0.3: reasons.append(f"Low credit utilization ({s.credit_utilization:.1%})")
    elif s.credit_utilization > 0.7: reasons.append(f"High credit utilization ({s.credit_utilization:.1%})")
    if not reasons: reasons.append("Transaction within normal parameters")
    return reasons[:4]


def build_leaderboard(s: TransactionState) -> List[LeaderboardEntry]:
    rewards = {a: round(random.uniform(-1.0, 1.0), 2) for a in ALL_ACTIONS}
    sorted_actions = sorted(rewards.items(), key=lambda x: x[1], reverse=True)
    return [LeaderboardEntry(action=a, reward=r) for a, r in sorted_actions[:5]]


def policy_checks(s: TransactionState, action: str) -> List[PolicyCheck]:
    checks = [
        PolicyCheck(name="Amount within limit", passed=s.amount <= s.available_credit, detail=f"Amount {s.amount} vs available {s.available_credit}"),
        PolicyCheck(name="Credit score threshold", passed=s.credit_score >= 600, detail=f"Score {s.credit_score} >= 600"),
        PolicyCheck(name="Recent payment activity", passed=(datetime.now() - datetime.fromisoformat(s.last_payment_date)).days < 45, detail=f"Last payment {s.last_payment_date}"),
        PolicyCheck(name="Account age minimum", passed=s.account_age_months >= 3, detail=f"Account age {s.account_age_months} months")
    ]
    return checks


def score_action(s: TransactionState, action: str) -> float:
    base = random.uniform(0.5, 1.0)
    if action == "approve_card_1" and s.credit_score >= 700: base += 0.3
    elif action == "escalate_review" and s.credit_score < 650: base += 0.3
    elif action == "deny_transaction" and s.credit_score < 600: base += 0.3
    return round(min(1.0, max(-1.0, base)), 2)


env = EnvStore()


@app.get("/")
def root():
    return {"service": "payment_credit_env", "status": "running", "version": "0.2.0"}


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/reset")
def reset():
    s = env.reset()
    risk_band = get_risk_band(s)
    rec = recommended_action_for_band(s, risk_band)
    return {"state": s, "risk_band": risk_band, "recommended_action": rec}


@app.get("/state")
def state():
    return env.get_state()


@app.post("/step", response_model=StepResponse)
def step(req: StepRequest):
    s = env.get_state()
    risk_band = get_risk_band(s)
    rec = recommended_action_for_band(s, risk_band)
    reasons = build_reasons(s, risk_band)
    leaderboard = build_leaderboard(s)
    checks = policy_checks(s, req.action)
    reward = score_action(s, req.action)
    env.audit_log.append(AuditLogEntry(
        timestamp=datetime.now().isoformat(),
        transaction_id=s.transaction_id,
        action=req.action,
        reward=reward,
        risk_band=risk_band,
        recommended_action=rec
    ))
    return StepResponse(
        transaction_id=s.transaction_id,
        action=req.action,
        reward=reward,
        done=True,
        state=s,
        risk_band=risk_band,
        recommended_action=rec,
        reasons=reasons,
        leaderboard=leaderboard,
        policy_checks=checks
    )


@app.get("/audit-log")
def get_audit_log():
    return env.audit_log