Spaces:
Sleeping
Sleeping
| # 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 | |
| ) | |
| # --------------------------------------------------- | |
| # 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 | |
| # --------------------------------------------------- | |
| # 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", | |
| "30+@9", | |
| "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" | |
| ) | |
| 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" | |
| ) | |
| output_table = gr.Dataframe() | |
| run_button.click( | |
| fn=run_analysis, | |
| inputs=[ | |
| dataset_dropdown, | |
| analysis_dropdown, | |
| category_dropdown | |
| ], | |
| outputs=output_table | |
| ) | |
| app.launch() |