Expense Tracker Chatbot is an innovative project that combines the power of natural language processing and data visualisation enabling users to track their expenses.
import pandas as pd
import requests
import json
import streamlit as st
import secrets
from streamlit_chat import message as st_message

st.title("Expense Tracker Assistant")

st.markdown(
        """
         <style>
            div[role=radiogroup] label:first-of-type {
            visibility: hidden;
            height: 0px;
            }
         </style>
        """,
        unsafe_allow_html=True,
   )


if "history" not in st.session_state:
    st.session_state.history = []

def generate_session_id():
    return secrets.token_hex(16)

def get_bot_response(message):
    url = 'http://localhost:5005/webhooks/rest/webhook'
    payload = {"sender": "user", "message": message}
    headers = {'Content-Type': 'application/json', 'Accept': 'text/plain'}
    response = requests.post(url, data=json.dumps(payload), headers=headers)
    return response.json()

def generate_response():

    if st.session_state.history != [] and "is_radio" in st.session_state.history[-1]:
        for key in st.session_state:
            if key.startswith("selected_option_"):
                option = st.session_state[key]
                if option == "default":
                    continue
                print(key, option)
                print("end")
                user_message = option
                break
    else:
        user_message = st.session_state.input_text
        
    bot_response = get_bot_response(user_message)
    st.session_state.history.append({"is_user": True, "message": user_message, "key" : generate_session_id()})
    for i in range(len(bot_response)):
        if "text" in bot_response[i]:
            message = bot_response[i]['text']
            st.session_state.history.append({"is_user": False, "message": message, "key" : generate_session_id()})
        if "buttons" in bot_response[i]:
            button_options = bot_response[i].get("buttons", [])
            options_list = [option["title"] for option in button_options]
            radio_session_id = generate_session_id()
            print("radio_session_id", radio_session_id)
            st.session_state.history.append({"is_user": False, "message": options_list, "key" : radio_session_id, "is_radio": True})
        if "image" in bot_response[i]:
            st.session_state.history.append({"is_user": False, "message": bot_response[i]["image"], "key" : generate_session_id(), "is_image": True})

for chat in st.session_state.history:
    if "is_radio" in chat:
        key = generate_session_id()
        st.radio("Select one of them", chat["message"], key=f"selected_option_{key}", on_change=generate_response)
    elif "is_image" in chat:
        data_file_path = chat["message"]
        df = pd.read_csv(data_file_path)
        x_axis_label = df.columns[0]
        st.bar_chart(df.set_index(x_axis_label))
    else:
        st_message(**chat)
st.text_input("Talk to the bot:", key="input_text", on_change=generate_response)


Expense-Tracker-Chatbot.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
# This files contains your custom actions which can be used to run
# custom Python code.
#
# See this guide on how to implement these action:
# https://rasa.com/docs/rasa/custom-actions


from typing import Any, Text, Dict, List
from pytz import timezone 
from datetime import datetime, timedelta

from rasa_sdk import Action, Tracker, FormValidationAction
from rasa_sdk.executor import CollectingDispatcher
from rasa_sdk.types import DomainDict
from rasa_sdk.events import AllSlotsReset
from database.AzureKeyVault import AzureKeyVault
from data_exporter.CSVFile import CSVFile
from database.SQLDatabase import SQLDatabase
from database.SQLQuery import SQLQuery
from database.DatabaseConfig import DatabaseConfig

EXPENSE_CATEGORIES =  ["household", "transportation", "work&education", "food&dining", "entertainment"]


class ValidateExpenseForm(FormValidationAction):
    def name(self) -> Text:
        return "validate_expense_form"

    def validate_expense_category(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.lower() in EXPENSE_CATEGORIES:
            dispatcher.utter_message(text=f"Great, I will categorize it as {slot_value}.")
            return {"expense_category": slot_value}
        elif slot_value.lower() == "default":
            dispatcher.utter_message(text=f"I don't understand the category")
            return {"expense_category": None}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter any of these 5 {'/'.join(EXPENSE_CATEGORIES)}.")
            return {"expense_category": None}
        
    def validate_expense_amount(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.isnumeric():
            dispatcher.utter_message(text=f"Sure, I will enter the amount as {slot_value}.")
            return {"expense_amount": slot_value}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter the amount in whole number?")
            return {"expense_amount": None}
        
    def validate_expense_date(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.lower() == "today":
            date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the date as {date}.")
            return {"expense_date": date}
        elif slot_value.lower() == "yesterday":
            date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=1)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the date as {date}.")
            return {"expense_date": date}
        elif slot_value.lower() == "day before yesterday":
            date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=2)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the date as {date}.")
            return {"expense_date": date}
        elif slot_value.lower() == "tomorrow":
            date = (datetime.now(timezone('Asia/Kolkata')) + timedelta(days=1)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the date as {date}.")
            return {"expense_date": date}
        elif slot_value.lower() == "day after tomorrow":
            date = (datetime.now(timezone('Asia/Kolkata')) + timedelta(days=2)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the date as {date}.")
            return {"expense_date": date}
        elif len(slot_value.split("-")) == 3:
            try:
                datetime.strptime(slot_value, '%Y-%m-%d')
                dispatcher.utter_message(text=f"Sure, I will enter the date as {slot_value}.")
                return {"expense_date": slot_value}
            except ValueError:
                dispatcher.utter_message(text=f"Sorry, Can you enter the date in yyyy-mm-dd format?")
                return {"expense_date": None}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter the date in yyyy-mm-dd format?")
            return {"expense_date": None}
        
    def validate_expense_confirmation(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if tracker.get_intent_of_latest_message() == "affirm":
            dispatcher.utter_message(text=f"Great, Your expense details are submitted.")
            return {"expense_confirmation": True}
        elif tracker.get_intent_of_latest_message() == "deny":
            dispatcher.utter_message(text=f"Ok, I will not submit the expense.")
            return {"expense_confirmation": False}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter either yes or no?")
            return {"expense_confirmation": None}

class AskForExpenseCategory(Action):
    def name(self) -> Text:
        return "action_ask_expense_category"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="Please select the category of your expense.", buttons=[{"payload": "default", "title": "default"}, {"payload": "household", "title": "household"}, {"payload": "transportation", "title": "transportation"}, {"payload": "work&education", "title": "work&education"}, {"payload": "food&dining", "title": "food&dining"}, {"payload": "entertainment", "title": "entertainment"}])
        return []
    
class ActionAskExpenseAmount(Action):
    def name(self) -> Text:
        return "action_ask_expense_amount"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="Please enter the expense amount in numbers.")
        return []
    
class ActionAskExpenseDate(Action):
    def name(self) -> Text:
        return "action_ask_expense_date"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="Please enter the date of your expenditure.")
        return []
    

class ActionResetExpenseForm(Action):
    def name(self) -> Text:
        return "action_reset_expense_form"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        return [AllSlotsReset()]

class ActionAskExpenseConfirmation(Action):
    def name(self) -> Text:
        return "action_ask_expense_confirmation"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        expense_category = tracker.get_slot("expense_category")
        expense_amount = tracker.get_slot("expense_amount")
        expense_date = tracker.get_slot("expense_date")
        message = f"I will submit the expense of {expense_amount} in {expense_category} category on {expense_date}. \n Please confirm."
        dispatcher.utter_message(text= message)
        return []
    
class ActionSubmitExpenseForm(Action):
    def name(self) -> Text:
        return "action_submit_expense_form"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        if tracker.get_intent_of_latest_message() == "affirm":
            expense_category = tracker.get_slot("expense_category")
            expense_amount = tracker.get_slot("expense_amount")
            expense_date = tracker.get_slot("expense_date")
            column_value_dict = {"category": expense_category, "amount": expense_amount, "date": expense_date}
            
            sql_server_name = DatabaseConfig.SQL_DATABASE_SERVER_NAME.value
            sql_database_name = DatabaseConfig.SQL_DATABASE_NAME.value
            sql_database_username = DatabaseConfig.SQL_DATABASE_USERNAME.value
            key_vault_url =  DatabaseConfig.KEY_VAULT_URL.value
            sql_database_password_secret_name = DatabaseConfig.SQL_DATABASE_PASSWORD_SECRET_NAME.value
            
            azureKeyVault = AzureKeyVault(key_vault_url)
            sql_database_password = azureKeyVault.get_secret(sql_database_password_secret_name)
            sqlDatabase = SQLDatabase(sql_server_name, sql_database_name, sql_database_username, sql_database_password)
            print(column_value_dict)
            sqlDatabase.insert_data("expense", column_value_dict)
            return []
        else:
            return [AllSlotsReset()]
        
class ActionAskVisulationGroupByType(Action):
    def name(self) -> Text:
        return "action_ask_vis_group_by_type"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="Should the visualisation show trends over time or should it focus on comparing expenses across different spending categories?.")
        return []   
    
class ActionAskVisualisationAggregateType(Action):
    def name(self) -> Text:
        return "action_ask_vis_aggregate_type"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="What kind of summary would you prefer for the expense amounts?", buttons=[{"payload": "default", "title": "default"}, {"payload": "sum", "title": "sum"}, {"payload": "average", "title": "average"}, {"payload": "minimum", "title": "minimum"}, {"payload": "maximum", "title": "maximum"}])
        return []
    
class ActionAskVisualisationTimePeriod(Action):
    def name(self) -> Text:
        return "action_ask_vis_time_period"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        dispatcher.utter_message(text="Could you please specify the specific timeframe you're interested in? Please provide the start and end dates for the period you'd like to analyze.")
        return []
    
class ActionSubmitVisualisationForm(Action):
    def name(self) -> Text:
        return "action_submit_visualisation_form"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        vis_group_by_type = tracker.get_slot("vis_group_by_type")
        vis_aggregate_type = tracker.get_slot("vis_aggregate_type")
        vis_time_period = tracker.get_slot("vis_time_period")
        message = f"Here is the visualisation of {vis_aggregate_type} of expenses over {vis_time_period} grouped by {vis_group_by_type}."

        sqlQuery = SQLQuery(vis_group_by_type, vis_aggregate_type, vis_time_period)
        query = sqlQuery.create()
        print(query)
        sql_server_name = DatabaseConfig.SQL_DATABASE_SERVER_NAME.value
        sql_database_name = DatabaseConfig.SQL_DATABASE_NAME.value
        sql_database_username = DatabaseConfig.SQL_DATABASE_USERNAME.value
        key_vault_url =  DatabaseConfig.KEY_VAULT_URL.value
        sql_database_password_secret_name = DatabaseConfig.SQL_DATABASE_PASSWORD_SECRET_NAME.value
        
        azureKeyVault = AzureKeyVault(key_vault_url)
        sql_database_password = azureKeyVault.get_secret(sql_database_password_secret_name)
        sqlDatabase = SQLDatabase(sql_server_name, sql_database_name, sql_database_username, sql_database_password)
        data = sqlDatabase.get_data(query)
        
        if vis_group_by_type in ["category", "categories"]:
            csvFile = CSVFile(data, ["Expense Categories", f"{vis_aggregate_type.capitalize()} Expense Amount"])
        else:
            csvFile = CSVFile(data, ["Time intervals", f"{vis_aggregate_type.capitalize()} Expense Amount"])
        data_path = csvFile.write()
        dispatcher.utter_message(text= message,image=data_path)
        return []
    
class ValidateVisualisationForm(FormValidationAction):
    def name(self) -> Text:
        return "validate_visualisation_form"

    def validate_vis_group_by_type(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.lower() in ["time", "categories", "category"]:
            dispatcher.utter_message(text=f"Great, I will group the expenses by {slot_value}.")
            return {"vis_group_by_type": slot_value}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter either time or category?")
            return {"vis_group_by_type": None}
        
    def validate_vis_aggregate_type(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.lower() in ["sum", "average", "minimum", "maximum"]:
            dispatcher.utter_message(text=f"Great, I will aggregate the expenses {slot_value}.")
            return {"vis_aggregate_type": slot_value}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter either sum, average, minimum or maximum?")
            return {"vis_aggregate_type": None}
        
    def validate_vis_time_period(self, slot_value: Any, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> Dict[Text, Any]:
        if slot_value.lower() == "last month":
            start_date = (datetime.now().replace(day=1) - timedelta(days=1)).replace(day=1).strftime('%Y-%m-%d')
            end_date = (datetime.now().replace(day=1) - timedelta(days=1)).strftime('%Y-%m-%d')
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}
        elif slot_value.lower() == "last week":
            start_date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=7)).strftime("%Y-%m-%d")
            end_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}
        elif slot_value.lower() == "this month":
            start_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-01")
            end_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}
        elif slot_value.lower() == "this week":
            start_date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=datetime.now(timezone('Asia/Kolkata')).weekday())).strftime("%Y-%m-%d")
            end_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}
        elif slot_value.lower() == "today":
            date = datetime.now(timezone('Asia/Kolkata')).strftime('%Y-%m-%d')
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {date} to {date}.")
            return {"vis_time_period": f"{date} to {date}"}
        elif slot_value.lower() == "yesterday":
            date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=1)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {date} to {date}.")
            return {"vis_time_period": f"{date} to {date}"}
        elif slot_value.lower() == "day before yesterday":
            date = (datetime.now(timezone('Asia/Kolkata')) - timedelta(days=2)).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {date} to {date}.")
            return {"vis_time_period": f"{date} to {date}"}
        elif slot_value.lower() == "last year":
            start_date = datetime(datetime.now().year - 1, 1, 1).strftime('%Y-%m-%d')
            end_date = (datetime(datetime.now().year, 1, 1) - timedelta(days=1)).strftime('%Y-%m-%d')
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}
        elif slot_value.lower() == "this year":
            start_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-01-01")
            end_date = datetime.now(timezone('Asia/Kolkata')).strftime("%Y-%m-%d")
            dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
            return {"vis_time_period": f"{start_date} to {end_date}"}        
        elif len(slot_value.split(" to ")) == 2:
            try:
                start_date, end_date = slot_value.split(" to ")
                datetime.strptime(start_date, "%Y-%m-%d")
                datetime.strptime(end_date, "%Y-%m-%d")
                dispatcher.utter_message(text=f"Sure, I will enter the time period as {start_date} to {end_date}.")
                return {"vis_time_period": f"{start_date} to {end_date}"}
            except ValueError:
                dispatcher.utter_message(text=f"Sorry, Can you enter the time period in yyyy-mm-dd to yyyy-mm-dd format?")
                return {"vis_time_period": None}
        else:
            dispatcher.utter_message(text=f"Sorry, Can you enter the time period in yyyy-mm-dd to yyyy-mm-dd format?")
            return {"vis_time_period": None}
        
class ActionResetVisualisationForm(Action):
    def name(self) -> Text:
        return "action_reset_visualisation_form"

    def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: DomainDict) -> List[Dict[Text, Any]]:
        return [AllSlotsReset()]

actions/actions.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
import csv
import uuid
import os

class CSVFile:

    def __init__(self, data, column_list):
        self.data = data
        self.column_list = column_list

    def generate_random_file_name(self):
        return str(uuid.uuid4()) + ".csv"
    
    def get_file_path(self):
        file_name = self.generate_random_file_name()
        try:
            os.mkdir(f"data_exporter/exports")
        except:
            print("Directory exports already exists")
            pass
        csv_file_path = f"data_exporter/exports/{file_name}"
        return csv_file_path
    
    def write(self):
        csv_file_path = self.get_file_path()
        with open(csv_file_path, "w", newline="") as csv_file:
            writer = csv.writer(csv_file)
            writer.writerow(self.column_list)
            writer.writerows(self.data)

        return csv_file_path
data_exporter/CSVFile.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient

class AzureKeyVault:

    def __init__(self, key_vault_url):
        self.key_vault_url = key_vault_url

    def get_secret(self, secret_name):
        credential = DefaultAzureCredential()
        secret_client = SecretClient(vault_url=self.key_vault_url, credential=credential)

        retrieved_secret = secret_client.get_secret(secret_name)
        return retrieved_secret.value

database/AzureKeyVault.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
from enum import Enum

class DatabaseConfig(Enum):

    SQL_DATABASE_SERVER_NAME = None
    SQL_DATABASE_NAME = None
    SQL_DATABASE_USERNAME = None
    KEY_VAULT_URL = None
    SQL_DATABASE_PASSWORD_SECRET_NAME = None
database/DatabaseConfig.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
import pyodbc
import pandas as pd

class SQLDatabase:

    def __init__(self, server, database, username, password):
        self.server = server
        self.database = database
        self.username = username
        self.password = password

    def connect(self):
        return pyodbc.connect(f"DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={self.server};DATABASE={self.database};UID={self.username};PWD={self.password}")
    
    def get_data(self, query):
        conn = self.connect()
        cursor = conn.cursor()
        cursor.execute(query)
        data = cursor.fetchall()
        conn.close()
        return data
    
    def get_data_frame(self, query):
        return pd.DataFrame(self.get_data(query))
    
    def insert_data(self, table_name, data):
        conn = self.connect()
        cursor = conn.cursor()
        columns = ", ".join(data.keys())
        values = ", ".join(["'" + str(value) + "'" for value in data.values()])
        sql_query = f"INSERT INTO {table_name} ({columns}) VALUES ({values})"
        cursor.execute(sql_query)
        conn.commit()
        conn.close()

database/SQLDatabase.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot
from datetime import datetime

class SQLQuery:

    def __init__(self, vis_group_by_type, vis_aggregate_type, vis_time_period):
        self.vis_group_by_type = vis_group_by_type
        self.vis_aggregate_type = vis_aggregate_type
        self.vis_time_period = vis_time_period

    def get_date_range(self):
        start_date, end_date = self.vis_time_period.split(' to ')
        start_date = datetime.strptime(start_date, '%Y-%m-%d')
        end_date = datetime.strptime(end_date, '%Y-%m-%d')
        return start_date, end_date
    
    def choose_grouping_strategy(self, start_date, end_date):
        interval_length = (end_date - start_date).days
        
        if interval_length <= 7:  
            return 'day'
        elif interval_length <= 30:  
            return 'week'
        elif interval_length <= 365: 
            return 'month'
        else:
            return 'year'
        
    def get_aggregate_type(self):
        aggregate_lookup = {"minimum": "MIN", "maximum": "MAX", "sum": "SUM", "average": "AVG"}
        return aggregate_lookup[self.vis_aggregate_type]
    
    def create(self):
        start_date, end_date = self.get_date_range()
        aggregate_type = self.get_aggregate_type()
        
        if self.vis_group_by_type == "category" or self.vis_group_by_type == "categories":
            sql_query = f"SELECT category, {aggregate_type}(amount) as {self.vis_aggregate_type}_amount FROM expense where date >= '{start_date.strftime('%Y-%m-%d')}' and date <= '{end_date.strftime('%Y-%m-%d')}' GROUP BY category"
        else:
            grouping_strategy = self.choose_grouping_strategy(start_date, end_date)
            if grouping_strategy == "week":
                sql_query = f"SELECT CONCAT('Week of ', MIN(date)) AS Week, {aggregate_type}(amount) as {self.vis_aggregate_type}_amount FROM expense where date >= '{start_date.strftime('%Y-%m-%d')}' and date <= '{end_date.strftime('%Y-%m-%d')}' GROUP BY DATEPART(WEEK, date)"
            elif grouping_strategy == "day":
                sql_query = f"SELECT date, {aggregate_type}(amount) as {self.vis_aggregate_type}_amount FROM expense where date >= '{start_date.strftime('%Y-%m-%d')}' and date <= '{end_date.strftime('%Y-%m-%d')}' GROUP BY date"
            elif grouping_strategy == "month":
                sql_query = f"SELECT DateName( month , DateAdd( month , DATEPART(MONTH, date) , -1 )) as Month, {aggregate_type}(amount) as {self.vis_aggregate_type}_amount FROM expense where date >= '{start_date.strftime('%Y-%m-%d')}' and date <= '{end_date.strftime('%Y-%m-%d')}' GROUP BY DATEPART(MONTH, date)"
            else:
                sql_query = f"SELECT DATEPART(YEAR, date) as Year, {aggregate_type}(amount) as {self.vis_aggregate_type}_amount FROM expense where date >= '{start_date.strftime('%Y-%m-%d')}' and date <= '{end_date.strftime('%Y-%m-%d')}' GROUP BY DATEPART(YEAR, date)"
        return sql_query
database/SQLQuery.py
Pallavi-Sinha-12/Expense-Tracker-Chatbot