Spaces:
Sleeping
Sleeping
File size: 2,617 Bytes
46bc116 b0bdce4 46bc116 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
# app.py
pip install -r requirements.txt
import gradio as gr
import pandas as pd
import numpy as np
import yfinance as yf
from ta import add_all_ta_features
from ta.utils import dropna
from transformers import pipeline, AutoModelForQuestionAnswering, AutoTokenizer
from sentence_transformers import SentenceTransformer
import faiss
# Load the dataset from Google Drive
def load_data():
url = "https://drive.google.com/uc?export=download&id=1N1bCVRSacs7_nENJzleqqNRHA22-H9B5"
df = pd.read_csv(url)
return df
# Preprocess the data
def preprocess_data(df):
df = dropna(df)
df = add_all_ta_features(df, open="Open", high="High", low="Low", close="Close", volume="Volume", fillna=True)
return df
# Train the FAISS index for RAG
def train_faiss_index(df):
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(df['Close'].astype(str).tolist())
dimension = embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(embeddings)
return index, model
# Load the QA model
def load_qa_model():
model_name = "deepset/roberta-base-squad2"
model = AutoModelForQuestionAnswering.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)
qa_pipeline = pipeline("question-answering", model=model, tokenizer=tokenizer)
return qa_pipeline
# Get technical analysis
def get_technical_analysis(df):
analysis = {
"SMA_50": df['Close'].rolling(window=50).mean().iloc[-1],
"SMA_200": df['Close'].rolling(window=200).mean().iloc[-1],
"RSI": df['momentum_rsi'].iloc[-1],
"MACD": df['trend_macd'].iloc[-1],
}
return analysis
# RAG-based QA function
def rag_qa(question, df, index, model, qa_pipeline):
query_embedding = model.encode([question])
distances, indices = index.search(query_embedding, k=1)
context = df.iloc[indices[0][0]]['Close']
result = qa_pipeline(question=question, context=str(context))
return result['answer']
# Gradio Interface
def trading_analysis_app(question):
df = load_data()
df = preprocess_data(df)
index, model = train_faiss_index(df)
qa_pipeline = load_qa_model()
analysis = get_technical_analysis(df)
answer = rag_qa(question, df, index, model, qa_pipeline)
return f"Technical Analysis: {analysis}\n\nAnswer: {answer}"
# Gradio Interface
iface = gr.Interface(
fn=trading_analysis_app,
inputs="text",
outputs="text",
title="RAG-based Trading Analysis",
description="Enter your question about ICICIBANK's stock to get technical analysis and answers."
)
iface.launch() |