appv2.py

# app.py

import pandas as pd
import gradio as gr

# ---------------------------------------------------
# HELPERS
# ---------------------------------------------------

from helper.vintage_helpers import (
    create_booking_vintage
)

from helper.data_merger import (
    merge_acq_perf
)

# ---------------------------------------------------
# METRICS
# ---------------------------------------------------

from metrics.mix_metrics import (
    calculate_vintage_mix,
    calculate_limit_mix
)

# ---------------------------------------------------
# ANALYTICS
# ---------------------------------------------------

from analytics.performance_analysis import (
    generate_metric_view
)

# ---------------------------------------------------
# VISUALIZATIONS - VINTAGE CURVES
# ---------------------------------------------------

from visualizations.vintage_curves import (
    generate_delinquency_metric_chart,
    generate_multi_metric_comparison,
    generate_segment_delinquency_curve
)

# ---------------------------------------------------
# VISUALIZATIONS - SEGMENT RANKING
# ---------------------------------------------------

from visualizations.segment_ranking import (
    generate_segment_risk_heatmap,
    generate_segment_risk_ranking,
    generate_multi_category_risk_comparison,
    calculate_portfolio_risk_summary
)

# ---------------------------------------------------
# LOAD DATA
# ---------------------------------------------------

acq = pd.read_csv(
    "data/acquisition.csv"
)

perf = pd.read_csv(
    "data/performance.csv"
)

# ---------------------------------------------------
# CREATE BOOKING VINTAGE
# ---------------------------------------------------

acq = create_booking_vintage(
    acq,
    booking_date_col="booking_date"
)

# ---------------------------------------------------
# CREATE MASTER PERFORMANCE DATASET
# ---------------------------------------------------

master_df = merge_acq_perf(
    acq_df=acq,
    perf_df=perf
)

# ---------------------------------------------------
# ACQUISITION ANALYSIS
# ---------------------------------------------------

def run_acquisition_analysis(
    analysis_type,
    category
):

    # -----------------------------------------
    # PORTFOLIO MIX
    # -----------------------------------------

    if analysis_type == "Portfolio Mix":

        result = (
            acq.groupby(
                ["booking_vintage", category]
            )
            .agg(
                count=("account_id", "nunique"),
                balance=("credit_limit", "sum")
            )
            .reset_index()
        )

        vintage_total = (
            result.groupby("booking_vintage")["count"]
            .transform("sum")
        )

        result["rate"] = (
            result["count"] / vintage_total
        ) * 100

        result["rate"] = (
            result["rate"]
            .round(2)
        )

    # -----------------------------------------
    # CREDIT LINE CONCENTRATION
    # -----------------------------------------

    elif analysis_type == "Credit Line Concentration":

        result = (
            acq.groupby(
                ["booking_vintage", category]
            )
            .agg(
                count=("account_id", "nunique"),
                balance=("credit_limit", "sum")
            )
            .reset_index()
        )

        vintage_total = (
            result.groupby("booking_vintage")["balance"]
            .transform("sum")
        )

        result["rate"] = (
            result["balance"] / vintage_total
        ) * 100

        result["rate"] = (
            result["rate"]
            .round(2)
        )

    else:

        return pd.DataFrame()

    # -----------------------------------------
    # STANDARDIZED OUTPUT
    # -----------------------------------------

    result = result.rename(
        columns={
            "booking_vintage": "Vintage",
            category: "Category",
            "count": "Count",
            "balance": "Balance",
            "rate": "Rate"
        }
    )

    return result[
        [
            "Vintage",
            "Category",
            "Count",
            "Balance",
            "Rate"
        ]
    ]

# ---------------------------------------------------
# PERFORMANCE ANALYSIS
# ---------------------------------------------------

def run_performance_analysis(
    metric_name,
    view_level
):

    # -----------------------------------------
    # VIEW MAPPING
    # -----------------------------------------

    view_mapping = {

        "Overall": None,

        "Channel":
        "sourcing_channel",

        "FICO":
        "fico_band",

        "City Tier":
        "city_tier",

        "Occupation":
        "occupation_type"
    }

    group_col = view_mapping[
        view_level
    ]

    # -----------------------------------------
    # CALL ANALYTICS ENGINE
    # -----------------------------------------

    result = generate_metric_view(
        df=master_df,
        metric_name=metric_name,
        group_col=group_col
    )

    # -----------------------------------------
    # STANDARDIZE OUTPUT
    # -----------------------------------------

    if group_col is not None:

        result = result.rename(
            columns={
                group_col: "Category"
            }
        )

    else:

        result["Category"] = "Overall"

    # -----------------------------------------
    # IDENTIFY RATE COLUMN
    # -----------------------------------------

    rate_col = [
        col for col in result.columns
        if "rate" in col.lower()
    ][0]

    # -----------------------------------------
    # OUTPUT FORMAT
    # -----------------------------------------

    final_result = pd.DataFrame()

    final_result["Vintage"] = (
        result["booking_vintage"]
    )

    final_result["Category"] = (
        result["Category"]
    )

    final_result["Count"] = (
        result["total_accounts"]
    )

    final_result["Balance"] = (
        result["total_balance"]
    )

    final_result["Rate"] = (
        result[rate_col]
        .round(2)
    )

    return final_result

# ---------------------------------------------------
# VINTAGE CURVES ANALYSIS
# ---------------------------------------------------

def generate_vintage_curve_single(
    metric_name
):
    """Generate single vintage curve for a metric."""
    try:
        fig = generate_delinquency_metric_chart(
            df=master_df,
            metric_name=metric_name,
            chart_type="line"
        )
        return fig
    except Exception as e:
        return f"Error generating vintage curve: {str(e)}"


def generate_vintage_curves_comparison():
    """Generate comparison of all vintage curves."""
    try:
        fig = generate_multi_metric_comparison(
            df=master_df,
            metrics=["30+@3", "30+@6", "60+@6", "Yr1 NCL"]
        )
        return fig
    except Exception as e:
        return f"Error generating comparison: {str(e)}"


def generate_segmented_vintage_curve(
    metric_name,
    category
):
    """Generate vintage curve segmented by category."""
    try:
        fig = generate_segment_delinquency_curve(
            df=master_df,
            metric_name=metric_name,
            category=category
        )
        return fig
    except Exception as e:
        return f"Error generating segmented curve: {str(e)}"

# ---------------------------------------------------
# SEGMENT RANKING ANALYSIS
# ---------------------------------------------------

def generate_segment_risk_heatmap_chart():
    """Generate risk heatmap across all segments and metrics."""
    try:
        fig = generate_segment_risk_heatmap(
            df=master_df
        )
        return fig
    except Exception as e:
        return f"Error generating heatmap: {str(e)}"


def generate_high_risk_segments_ranking(
    metric_name,
    category
):
    """Generate ranking of high-risk segments."""
    try:
        fig = generate_segment_risk_ranking(
            df=master_df,
            metric_name=metric_name,
            category=category,
            top_n=10
        )
        return fig
    except Exception as e:
        return f"Error generating ranking: {str(e)}"


def generate_multi_category_comparison(
    metric_name
):
    """Generate risk comparison across all categories."""
    try:
        fig = generate_multi_category_risk_comparison(
            df=master_df,
            metric_name=metric_name
        )
        return fig
    except Exception as e:
        return f"Error generating comparison: {str(e)}"


def generate_portfolio_summary():
    """Generate portfolio risk summary."""
    try:
        summary_df = calculate_portfolio_risk_summary(
            df=master_df
        )
        return summary_df
    except Exception as e:
        return f"Error generating summary: {str(e)}"

# ---------------------------------------------------
# DYNAMIC DROPDOWNS
# ---------------------------------------------------

def update_analysis_dropdown(
    dataset
):

    # -----------------------------------------
    # ACQUISITION
    # -----------------------------------------

    if dataset == "Acquisition":

        return gr.update(
            choices=[
                "Portfolio Mix",
                "Credit Line Concentration"
            ],
            value="Portfolio Mix"
        )

    # -----------------------------------------
    # PERFORMANCE
    # -----------------------------------------

    elif dataset == "Performance":

        return gr.update(
            choices=[
                "30+@3",
                "30+@6",
                "60+@6",
                "Yr1 NCL"
            ],
            value="30+@6"
        )


def update_category_dropdown(
    dataset
):

    # -----------------------------------------
    # ACQUISITION
    # -----------------------------------------

    if dataset == "Acquisition":

        return gr.update(
            choices=[
                "fico_band",
                "sourcing_channel",
                "city_tier",
                "occupation_type"
            ],
            value="fico_band"
        )

    # -----------------------------------------
    # PERFORMANCE
    # -----------------------------------------

    elif dataset == "Performance":

        return gr.update(
            choices=[
                "Overall",
                "Channel",
                "FICO",
                "City Tier",
                "Occupation"
            ],
            value="Overall"
        )

# ---------------------------------------------------
# MASTER ROUTER
# ---------------------------------------------------

def run_analysis(
    dataset,
    analysis,
    category
):

    # -----------------------------------------
    # ACQUISITION
    # -----------------------------------------

    if dataset == "Acquisition":

        return run_acquisition_analysis(
            analysis_type=analysis,
            category=category
        )

    # -----------------------------------------
    # PERFORMANCE
    # -----------------------------------------

    elif dataset == "Performance":

        return run_performance_analysis(
            metric_name=analysis,
            view_level=category
        )

    else:

        return pd.DataFrame()

# ---------------------------------------------------
# GRADIO UI
# ---------------------------------------------------

with gr.Blocks() as app:

    gr.Markdown(
        "# Risk Analytics Manager Agent - Phase 2"
    )

    with gr.Tabs():

        # =================================================
        # TAB 1: BASIC ANALYSIS (Phase 1)
        # =================================================

        with gr.TabItem("📊 Basic Analysis"):

            gr.Markdown(
                "## Phase 1: Acquisition & Performance Analysis"
            )

            with gr.Row():

                dataset_dropdown = gr.Dropdown(
                    choices=[
                        "Acquisition",
                        "Performance"
                    ],
                    value="Acquisition",
                    label="Dataset"
                )

                analysis_dropdown = gr.Dropdown(
                    choices=[
                        "Portfolio Mix",
                        "Credit Line Concentration"
                    ],
                    value="Portfolio Mix",
                    label="Analysis"
                )

                category_dropdown = gr.Dropdown(
                    choices=[
                        "fico_band",
                        "sourcing_channel",
                        "city_tier",
                        "occupation_type"
                    ],
                    value="fico_band",
                    label="Category / View"
                )

            # -----------------------------------------
            # DYNAMIC DROPDOWNS
            # -----------------------------------------

            dataset_dropdown.change(
                fn=update_analysis_dropdown,
                inputs=dataset_dropdown,
                outputs=analysis_dropdown
            )

            dataset_dropdown.change(
                fn=update_category_dropdown,
                inputs=dataset_dropdown,
                outputs=category_dropdown
            )

            # -----------------------------------------
            # RUN BUTTON
            # -----------------------------------------

            run_button = gr.Button(
                "Run Analysis",
                variant="primary"
            )

            output_table = gr.Dataframe()

            run_button.click(
                fn=run_analysis,
                inputs=[
                    dataset_dropdown,
                    analysis_dropdown,
                    category_dropdown
                ],
                outputs=output_table
            )

        # =================================================
        # TAB 2: VINTAGE CURVES (Phase 2)
        # =================================================

        with gr.TabItem("📈 Vintage Curves"):

            gr.Markdown(
                "## Phase 2: Vintage Delinquency Curves Analysis"
            )

            with gr.Row():

                metric_dropdown = gr.Dropdown(
                    choices=[
                        "30+@3",
                        "30+@6",
                        "60+@6",
                        "Yr1 NCL"
                    ],
                    value="30+@6",
                    label="Delinquency Metric"
                )

                vintage_chart_type = gr.Radio(
                    choices=["Single Metric", "All Metrics Comparison"],
                    value="Single Metric",
                    label="Chart Type"
                )

            def update_vintage_view(metric, chart_type):
                if chart_type == "Single Metric":
                    return generate_vintage_curve_single(metric)
                else:
                    return generate_vintage_curves_comparison()

            vintage_chart = gr.Plot(
                label="Vintage Curve"
            )

            gen_vintage_btn = gr.Button(
                "Generate Vintage Curve",
                variant="primary"
            )

            gen_vintage_btn.click(
                fn=update_vintage_view,
                inputs=[metric_dropdown, vintage_chart_type],
                outputs=vintage_chart
            )

            gr.Markdown(
                "### Segmented Vintage Curves"
            )

            with gr.Row():

                segment_metric = gr.Dropdown(
                    choices=[
                        "30+@3",
                        "30+@6",
                        "60+@6",
                        "Yr1 NCL"
                    ],
                    value="30+@6",
                    label="Metric"
                )

                segment_category = gr.Dropdown(
                    choices=[
                        "fico_band",
                        "sourcing_channel",
                        "city_tier",
                        "occupation_type"
                    ],
                    value="fico_band",
                    label="Category"
                )

            segmented_chart = gr.Plot(
                label="Segmented Vintage Curve"
            )

            gen_segment_btn = gr.Button(
                "Generate Segmented Curve",
                variant="primary"
            )

            gen_segment_btn.click(
                fn=generate_segmented_vintage_curve,
                inputs=[segment_metric, segment_category],
                outputs=segmented_chart
            )

        # =================================================
        # TAB 3: SEGMENT RANKING (Phase 2)
        # =================================================

        with gr.TabItem("⚠️ Segment Ranking"):

            gr.Markdown(
                "## Phase 2: High-Risk Segment Analysis"
            )

            # --------- HEATMAP SECTION ---------

            gr.Markdown(
                "### 🔥 Overall Risk Heatmap"
            )

            gr.Markdown(
                "Risk scores across all delinquency metrics and segments"
            )

            heatmap_chart = gr.Plot(
                label="Risk Heatmap"
            )

            gen_heatmap_btn = gr.Button(
                "Generate Risk Heatmap",
                variant="primary"
            )

            gen_heatmap_btn.click(
                fn=generate_segment_risk_heatmap_chart,
                outputs=heatmap_chart
            )

            gr.Markdown(
                "---"
            )

            # --------- HIGH-RISK RANKING SECTION ---------

            gr.Markdown(
                "### 📊 High-Risk Segments Ranking"
            )

            with gr.Row():

                ranking_metric = gr.Dropdown(
                    choices=[
                        "30+@3",
                        "30+@6",
                        "60+@6",
                        "Yr1 NCL"
                    ],
                    value="30+@6",
                    label="Metric"
                )

                ranking_category = gr.Dropdown(
                    choices=[
                        "fico_band",
                        "sourcing_channel",
                        "city_tier",
                        "occupation_type"
                    ],
                    value="fico_band",
                    label="Category"
                )

            ranking_chart = gr.Plot(
                label="High-Risk Segments"
            )

            gen_ranking_btn = gr.Button(
                "Generate Risk Ranking",
                variant="primary"
            )

            gen_ranking_btn.click(
                fn=generate_high_risk_segments_ranking,
                inputs=[ranking_metric, ranking_category],
                outputs=ranking_chart
            )

            gr.Markdown(
                "---"
            )

            # --------- MULTI-CATEGORY COMPARISON ---------

            gr.Markdown(
                "### 🔀 Cross-Category Risk Comparison"
            )

            comparison_metric = gr.Dropdown(
                choices=[
                    "30+@3",
                    "30+@6",
                    "60+@6",
                    "Yr1 NCL"
                ],
                value="30+@6",
                label="Metric"
            )

            comparison_chart = gr.Plot(
                label="Multi-Category Comparison"
            )

            gen_comparison_btn = gr.Button(
                "Generate Comparison",
                variant="primary"
            )

            gen_comparison_btn.click(
                fn=generate_multi_category_comparison,
                inputs=comparison_metric,
                outputs=comparison_chart
            )

            gr.Markdown(
                "---"
            )

            # --------- PORTFOLIO SUMMARY ---------

            gr.Markdown(
                "### 📋 Portfolio Risk Summary"
            )

            summary_table = gr.Dataframe(
                label="Risk Summary"
            )

            gen_summary_btn = gr.Button(
                "Generate Summary",
                variant="primary"
            )

            gen_summary_btn.click(
                fn=generate_portfolio_summary,
                outputs=summary_table
            )

app.launch()