app.py

import locale
import gradio as gr
import matplotlib.pyplot as plt
import numpy as np
import numpy_financial as npf
import pandas as pd
import plotly.graph_objects as go
import seaborn as sns
from pandas.tseries.offsets import DateOffset, MonthEnd
from scipy import optimize

from portfolio import calculate_portfolio_returns
from utils import get_all_mf_schemes_df,get_mf_scheme_data

js_func = """
function refresh() {
    const url = new URL(window.location);

    if (url.searchParams.get('__theme') !== 'dark') {
        url.searchParams.set('__theme', 'dark');
        window.location.href = url.href;
    }
}
"""

locale.setlocale(locale.LC_MONETARY, 'en_IN')

def get_portfolio_report(*args):
    period = args[0]
    custom_start_date = args[1]
    custom_end_date = args[2]
    SIP_Date = args[3]
    sip_amount = args[4]
    lumpsum_amount = args[5]
    stepup = args[6]
    schemes_df = args[7]

    # Extract scheme names and weights, into a dictionary from the args
    scheme_name_and_weight = {}
    for i in range(8, len(args) - 1, 2):
        if args[i] and args[i+1]:
            scheme_name_and_weight[args[i]] = float(args[i+1])

    use_inception_date = args[-1]

    if not scheme_name_and_weight:
        return "Please add at least one scheme.", None, None, None
    
    end_date = pd.Timestamp.now().floor('D')

    if use_inception_date:
        start_date = pd.Timestamp(custom_start_date)
    elif period == "Custom":
        if not custom_start_date or not custom_end_date:
            return "Please provide both start and end dates for custom period.", None, None, None
        start_date = pd.Timestamp(custom_start_date)
        end_date = pd.Timestamp(custom_end_date)
    elif period == "YTD":
        start_date = pd.Timestamp(f"{end_date.year}-01-01")
    elif not period:
        return "Please select a period, provide custom dates, or use the inception date.", None, None, None
    else:
        period_parts = period.split()
        if len(period_parts) < 2:
            return "Invalid period selected.", None, None, None
        
        if 'year' in period_parts[1]:
            years = int(period_parts[0])
            start_date = end_date - DateOffset(years=years)
        else:
            months = int(period_parts[0])
            start_date = end_date - DateOffset(months=months)

    portfolio_report_string = calculate_portfolio_returns(scheme_name_and_weight, sip_amount, lumpsum_amount, stepup, start_date, end_date, SIP_Date, schemes_df)
    return portfolio_report_string


def quick_search_schemes(query, schemes_df):
    if not query:
        return []
    matching_schemes = schemes_df[schemes_df['schemeName'].str.contains(query, case=False, na=False)]
    return matching_schemes['schemeName'].tolist()[:40]

def update_scheme_dropdown(query, schemes_df, key_up_data: gr.KeyUpData):
    schemes = quick_search_schemes(key_up_data.input_value, schemes_df)
    return gr.update(choices=schemes, visible=True)

def update_schemes_list(schemes_list, updated_data):
    new_schemes_list = []
    for _, row in updated_data.iterrows():
        scheme_name = row.get('Scheme Name')
        weight = row.get('Weight (%)')
        action = row.get('Actions')
        if scheme_name and weight is not None and action != '🗑️':  # Only keep rows that aren't marked for deletion
            try:
                weight_float = float(weight)
                new_schemes_list.append((scheme_name, weight_float))
            except ValueError:
                # If weight is not a valid float, skip this row
                continue
    return new_schemes_list

def update_schemes_table(schemes_list):
    df = pd.DataFrame(schemes_list, columns=["Scheme Name", "Weight (%)"])
    df["Actions"] = "❌"
    
    # Calculate the sum of weights
    total_weight = df["Weight (%)"].sum()
    
    # Add a row for the total
    total_row = pd.DataFrame({
        "Scheme Name": ["Total"],
        "Weight (%)": [total_weight],
        "Actions": [""]
    })
    
    # Concatenate the original dataframe with the total row
    df = pd.concat([df, total_row], ignore_index=True)
    
    # Add a warning if total weight exceeds 100%
    if total_weight > 100:
        df.loc[df.index[-1], "Actions"] = "⚠️ Exceeds 100%"
    
    return df

def add_scheme_to_list(schemes_list, scheme_name, weight):
    if scheme_name and weight:
        new_list = schemes_list + [(scheme_name, float(weight))]
        return new_list, update_schemes_table(new_list), None, 0
    return schemes_list, update_schemes_table(schemes_list), scheme_name, weight

def update_schemes(schemes_list, updated_data):
    try:
        new_schemes_list = []
        for _, row in updated_data.iterrows():
            scheme_name = row.get('Scheme Name')
            weight = row.get('Weight (%)')
            if scheme_name != 'Total' and weight is not None:
                try:
                    weight_float = float(weight)
                    new_schemes_list.append((scheme_name, weight_float))
                except ValueError:
                    continue
        if not new_schemes_list:
            return schemes_list, update_schemes_table(schemes_list), "No valid schemes found in the table."
        return new_schemes_list, update_schemes_table(new_schemes_list), None
    except Exception as e:
        error_msg = f"Error updating schemes: {str(e)}"
        return schemes_list, update_schemes_table(schemes_list), error_msg

def prepare_inputs(period, custom_start, custom_end, SIP_Date, sip_amount, schemes_list, schemes_df):
    inputs = [period, custom_start, custom_end, SIP_Date, sip_amount, schemes_df]
    for name, weight in schemes_list:
        inputs.extend([name, weight])
    return inputs

def handle_row_selection(schemes_list, evt: gr.SelectData, table_data):
    if evt.index is not None and len(evt.index) > 1:
        column_index = evt.index[1]
        if column_index == 2:  # "Actions" column
            row_index = evt.index[0]
            if row_index < len(table_data) - 1:  # Ensure we're not trying to delete the total row
                # Remove the row
                table_data = table_data.drop(row_index).reset_index(drop=True)
                # Update the schemes_list
                updated_schemes_list = [(row['Scheme Name'], row['Weight (%)']) for _, row in table_data.iterrows() if row['Scheme Name'] != 'Total']
                # Recalculate the total
                return update_schemes_table(updated_schemes_list), updated_schemes_list
    return table_data, schemes_list

def create_ui():
    schemes_df = get_all_mf_schemes_df()
    with gr.Blocks(js=js_func) as demo:
        gr.Markdown("# Mutual Fund SIP Returns Calculator")

        with gr.Row():
            period = gr.Dropdown(choices=["YTD", "1 month","3 months","6 months","1 year", "3 years", "5 years", "7 years", "10 years","15 years","20 years", "Custom"], label="Select Period",value="YTD")
            custom_start_date = gr.Textbox(label="Custom Start Date (YYYY-MM-DD)", visible=False)
            custom_end_date = gr.Textbox(label="Custom End Date (YYYY-MM-DD)", visible=False)
            SIP_Date = gr.Dropdown(label="Monthly SIP Date", choices=["start","middle","end"],value="start")
            with gr.Column():
                use_inception_date = gr.Checkbox(label="Use Earliest Inception Date", value=False)
                inception_date_display = gr.Textbox(label="Earliest Inception Date", interactive=False)

        with gr.Row():
            sip_amount = gr.Number(label="SIP Amount (₹)")
            upfront_amount = gr.Number(label="Upfront Investment (₹)",value=0)
            stepup = gr.Number(label="Stepup %",value=0)

        schemes_list = gr.State([])
        
        with gr.Row():
            scheme_dropdown = gr.Dropdown(label="Select Scheme", choices=[], allow_custom_value=True, interactive=True)
            scheme_weight = gr.Slider(minimum=0, maximum=100, step=1, label="Scheme Weight (%)")
            add_button = gr.Button("Add Scheme")

        schemes_table = gr.Dataframe(
            headers=["Scheme Name", "Weight (%)", "Actions"],
            datatype=["str", "number", "str"],
            col_count=(3, "fixed"),
            label="Added Schemes",
            type="pandas",
            interactive=True
        )

        update_button = gr.Button("Update Schemes")
        error_message = gr.Textbox(label="Error", visible=False)

        calculate_button = gr.Button("Calculate Returns")
        
        result = gr.Textbox(label="Results",)
        # pie_chart = gr.Plot(label="Scheme Weightages")
        # final_value = gr.Number(label="Final Value (₹)", interactive=False)
        # total_investment = gr.Number(label="Total Investment (₹)", interactive=False)

        def update_custom_date_visibility(period):
            return {custom_start_date: gr.update(visible=period=="Custom"),
                    custom_end_date: gr.update(visible=period=="Custom")}

        period.change(update_custom_date_visibility, inputs=[period], outputs=[custom_start_date, custom_end_date])

        scheme_dropdown.key_up(
            fn=update_scheme_dropdown,
            inputs=[scheme_dropdown, gr.State(schemes_df)],
            outputs=scheme_dropdown,
            queue=False,
            show_progress="hidden"
        )

        add_button.click(add_scheme_to_list, 
                         inputs=[schemes_list, scheme_dropdown, scheme_weight], 
                         outputs=[schemes_list, schemes_table, scheme_dropdown, scheme_weight])

        def update_schemes_and_show_error(schemes_list, updated_data):
            new_schemes_list, updated_table, error = update_schemes(schemes_list, updated_data)
            return (
                new_schemes_list,
                updated_table,
                gr.update(value=error, visible=bool(error))
            )

        update_button.click(
            update_schemes_and_show_error,
            inputs=[schemes_list, schemes_table],
            outputs=[schemes_list, schemes_table, error_message]
        )

        schemes_table.select(
                handle_row_selection,
                inputs=[schemes_list, schemes_table],
                outputs=[schemes_table, schemes_list]
        )

        def get_earliest_inception_date(schemes_list, schemes_df):
            inception_dates = []
            for scheme_name, _ in schemes_list:
                scheme_code = schemes_df[schemes_df['schemeName'] == scheme_name]['schemeCode'].values[0]
                _, inception_date = get_mf_scheme_data(scheme_code)
                inception_dates.append(inception_date)
            return max(inception_dates).strftime("%Y-%m-%d") if inception_dates else ""

        def update_inception_date(use_inception_date, schemes_list, schemes_df):
            if use_inception_date and schemes_list:
                earliest_inception_date = get_earliest_inception_date(schemes_list, schemes_df)
                return gr.update(value=earliest_inception_date, visible=True)
            else:
                return gr.update(value="", visible=False)

        use_inception_date.change(
            update_inception_date,
            inputs=[use_inception_date, schemes_list, gr.State(schemes_df)],
            outputs=inception_date_display
        )

        def prepare_inputs_with_inception(period, custom_start, custom_end, SIP_Date, sip_amount, upfront_amount,stepup, schemes_list, schemes_df, use_inception_date, inception_date_display):
            inputs = [period, custom_start, custom_end, SIP_Date, sip_amount, upfront_amount, stepup, schemes_df]
            for name, weight in schemes_list:
                inputs.extend([name, weight])
            
            inputs.append(use_inception_date)  # Add use_inception_date to the inputs
            if use_inception_date and inception_date_display:
                inputs[1] = inception_date_display  # Replace custom_start with inception_date_display
            
            return inputs

        calculate_button.click(
            lambda *args: get_portfolio_report(*prepare_inputs_with_inception(*args)),
            inputs=[period, custom_start_date, custom_end_date, SIP_Date, sip_amount,upfront_amount,stepup,schemes_list, gr.State(schemes_df), use_inception_date, inception_date_display],
            outputs=[result]
            # outputs=[result, final_value, total_investment]
            # outputs=[result, pie_chart, final_value, total_investment]
        )

    return demo

demo = create_ui()
demo.launch(debug=True)