Create app.py

app.py

@@ -0,0 +1,124 @@
+import gradio as gr
+from openai import OpenAI
+import snowflake.connector
+import os
+import json
+from decimal import Decimal
+from datetime import date, datetime
+
+
+# Initialize OpenAI client
+client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
+
+def fetch_trades_of_the_day():
+    """
+    Fetches RSI data from Snowflake, calculates future return percentages based on a lead value,
+    and returns the data as a formatted JSON string.
+
+    Parameters:
+    - horizon (int): Number of days to use in the lead function to calculate future return. Default is 21.
+
+    Returns:
+    - json_data (str): Formatted JSON string containing the RSI data.
+    """
+
+    def custom_json_serializer(obj):
+        """ Custom JSON serializer for handling date objects and Decimal types """
+        if isinstance(obj, (datetime, date)):
+            return obj.isoformat()  # Convert date/datetime to ISO format
+        elif isinstance(obj, Decimal):
+            return float(obj)  # Convert Decimal to float
+        raise TypeError(f"Type {type(obj)} not serializable")
+
+    try:
+        # Establish connection to Snowflake
+        conn = snowflake.connector.connect(
+            user=userdata.get('SNOWFLAKE_USER'),
+            password=userdata.get('SNOWFLAKE_PW'),
+            account=userdata.get('SNOWFLAKE_ACCOUNT'),
+            warehouse=userdata.get('SNOWFLAKE_WH'),
+            database=userdata.get('SNOWFLAKE_DB'),
+            schema='RESEARCHDATA'
+        )
+
+        # Define the query
+        query = """SELECT  
+                          TICKER, 
+                          RSI_CURRENT,
+                          RSI_CONDITION,
+                          CONSECUTIVE_DAYS_IN_RSI_CONDITION,
+                          HISTORICAL_OCCURRENCES,
+                          MEDIAN_FORWARD_5DAY_RETURN_PERCENT,
+                          AVG_FORWARD_5DAY_RETURN_PERCENT,
+                          POSITIVE_RATE, 
+                          NEGATIVE_RATE 
+                  
+                  FROM researchdata.RSI_TRADE_OF_THE_DAY 
+           
+                  ORDER BY trade_rank;"""
+
+        # Execute the query and fetch data
+        cur = conn.cursor()
+        rows = cur.execute(query).fetchall()
+        columns = [desc[0] for desc in cur.description]  # Get column names
+
+        # Close the cursor and connection
+        cur.close()
+        conn.close()
+
+        # Convert the rows into a list of dictionaries (for JSON serialization)
+        result = [dict(zip(columns, row)) for row in rows]
+
+        # Convert the result to a formatted JSON string, with the custom serializer
+        json_data = json.dumps(result, indent=4, default=custom_json_serializer)
+
+        return json_data
+
+    except Exception as e:
+        print(f"Failed to connect to Snowflake: {e}")
+        return None
+
+# Function to interact with the OpenAI assistant
+def interact_with_assistant(user_input):
+    thread = client.beta.threads.create()
+    
+    client.beta.threads.messages.create(
+        thread_id=thread.id,
+        role="user",
+        content=user_input,
+    )
+
+    run = client.beta.threads.runs.create(
+        thread_id=thread.id,
+        assistant_id='asst_WeygFo0phjmDVfjzyM9zktgl',  # Replace with your actual assistant ID
+    )
+
+    while run.status != 'completed':
+        run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
+        
+        if run.status == 'requires_action':
+            tool_outputs = []
+            for tool_call in run.required_action.submit_tool_outputs.tool_calls:
+                if tool_call.function.name == "fetch_trades_of_the_day":
+                    output = fetch_trades_of_the_day()
+                    tool_outputs.append({"tool_call_id": tool_call.id, "output": output})
+            
+            client.beta.threads.runs.submit_tool_outputs(
+                thread_id=thread.id,
+                run_id=run.id,
+                tool_outputs=tool_outputs
+            )
+
+    messages = client.beta.threads.messages.list(thread_id=thread.id)
+    return messages.data[0].content[0].text.value
+
+# Gradio interface
+iface = gr.Interface(
+    fn=interact_with_assistant,
+    inputs="text",
+    outputs="text",
+    title="Stock Market Assistant",
+    description="Ask questions about the stock market and get insights based on RSI data."
+)
+
+iface.launch()