Spaces:
Running
Running
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| import pandas as pd | |
| # Theme settings | |
| BG_COLOR = "#0f1117" | |
| FONT_COLOR = "white" | |
| COLOR_HIGH = "#e63946" | |
| COLOR_MED = "#f4a261" | |
| COLOR_LOW = "#2a9d8f" | |
| COLOR_MAP = { | |
| "High": COLOR_HIGH, | |
| "Medium": COLOR_MED, | |
| "Low": COLOR_LOW | |
| } | |
| def apply_theme(fig): | |
| fig.update_layout( | |
| paper_bgcolor=BG_COLOR, | |
| plot_bgcolor=BG_COLOR, | |
| font_color=FONT_COLOR, | |
| margin=dict(l=40, r=40, t=40, b=40) | |
| ) | |
| return fig | |
| def risk_distribution_chart(df): | |
| counts = df['risk_level'].value_counts().reset_index() | |
| counts.columns = ['risk_level', 'count'] | |
| fig = px.pie(counts, values='count', names='risk_level', color='risk_level', | |
| color_discrete_map=COLOR_MAP, hole=0.4) | |
| fig.update_traces(textposition='inside', textinfo='percent+label') | |
| return apply_theme(fig) | |
| def flagged_transactions_timeline(df): | |
| flagged_df = df[df['is_flagged'] == 1].copy() | |
| if flagged_df.empty: | |
| return px.line(title="No Flagged Transactions") | |
| flagged_df['date'] = flagged_df['timestamp'].dt.date | |
| daily_counts = flagged_df.groupby('date').size().reset_index(name='count') | |
| fig = px.line(daily_counts, x='date', y='count', markers=True) | |
| fig.update_traces(line_color=COLOR_HIGH) | |
| return apply_theme(fig) | |
| def amount_vs_risk_scatter(df): | |
| # size needs to be positive | |
| df['size'] = df['transaction_velocity'].clip(lower=1) | |
| # create a rule string for hover | |
| df['rule_str'] = df['rule_flags'].apply(lambda x: ", ".join(x) if isinstance(x, list) and x else "None") | |
| fig = px.scatter(df, x='amount', y='risk_score', color='risk_level', | |
| size='size', hover_data=['transaction_id', 'rule_str'], | |
| color_discrete_map=COLOR_MAP, log_x=True) | |
| return apply_theme(fig) | |
| def transaction_type_breakdown(df): | |
| grouped = df.groupby(['transaction_type', 'is_flagged']).size().reset_index(name='count') | |
| grouped['Status'] = grouped['is_flagged'].map({1: 'Flagged', 0: 'Clean'}) | |
| fig = px.bar(grouped, x='transaction_type', y='count', color='Status', barmode='group', | |
| color_discrete_map={'Flagged': COLOR_HIGH, 'Clean': COLOR_LOW}) | |
| return apply_theme(fig) | |
| def top_flagged_customers_chart(df): | |
| flagged = df[df['is_flagged'] == 1] | |
| if flagged.empty: | |
| return px.bar(title="No Flagged Customers") | |
| cust_stats = flagged.groupby('customer_id').agg( | |
| flagged_count=('transaction_id', 'count'), | |
| avg_risk=('risk_score', 'mean') | |
| ).reset_index().sort_values('flagged_count', ascending=False).head(10) | |
| fig = px.bar(cust_stats, x='flagged_count', y='customer_id', orientation='h', | |
| color='avg_risk', color_continuous_scale='Reds') | |
| fig.update_layout(yaxis={'categoryorder':'total ascending'}) | |
| return apply_theme(fig) | |
| def kyc_tier_distribution(profile_df): | |
| counts = profile_df['kyc_tier'].value_counts().reset_index() | |
| counts.columns = ['kyc_tier', 'count'] | |
| fig = px.pie(counts, values='count', names='kyc_tier', color='kyc_tier', | |
| color_discrete_map=COLOR_MAP, hole=0.5) | |
| return apply_theme(fig) | |
| def rule_trigger_frequency(df): | |
| all_rules = [] | |
| for rules in df['rule_flags']: | |
| if isinstance(rules, list): | |
| all_rules.extend(rules) | |
| if not all_rules: | |
| return px.bar(title="No Rules Triggered") | |
| counts = pd.Series(all_rules).value_counts().reset_index() | |
| counts.columns = ['Rule', 'Count'] | |
| fig = px.bar(counts, x='Count', y='Rule', orientation='h', color_discrete_sequence=[COLOR_MED]) | |
| fig.update_layout(yaxis={'categoryorder':'total ascending'}) | |
| return apply_theme(fig) | |