Update app.py

app.py

@@ -22,93 +22,217 @@ import json
 # Initialize Groq Client
 client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
 
-# ---------------------- Data Acquisition Layer ---------------------------
-class DataSource(ABC):
-    """Base class for data sources."""
+# ---------------------- Base Classes and Schemas ---------------------------
+class ResearchInput(BaseModel):
+    """Base schema for research tool inputs"""
+    data_key: str = Field(..., description="Session state key containing DataFrame")
+    columns: Optional[List[str]] = Field(None, description="List of columns to analyze")
+
+class TemporalAnalysisInput(ResearchInput):
+    """Schema for temporal analysis"""
+    time_col: str = Field(..., description="Name of timestamp column")
+    value_col: str = Field(..., description="Name of value column to analyze")
+
+class HypothesisInput(ResearchInput):
+    """Schema for hypothesis testing"""
+    group_col: str = Field(..., description="Categorical column defining groups")
+    value_col: str = Field(..., description="Numerical column to compare")
+
+class ModelTrainingInput(ResearchInput):
+    """Schema for model training"""
+    target_col: str = Field(..., description="Name of target column")
+
+class DataAnalyzer(ABC):
+    """Abstract base class for data analysis modules"""
     @abstractmethod
-    def connect(self) -> None:
-        """Connect to the data source."""
+    def invoke(self, **kwargs) -> Dict[str, Any]:
         pass
-    
-    @abstractmethod
-    def fetch_data(self, query: str, **kwargs) -> pd.DataFrame:
-         """Fetch the data based on a specific query."""
-         pass
 
+# ---------------------- Concrete Analyzer Implementations ---------------------------
+class AdvancedEDA(DataAnalyzer):
+    """Comprehensive Exploratory Data Analysis"""
+    def invoke(self, data: pd.DataFrame, **kwargs) -> Dict[str, Any]:
+        try:
+            analysis = {
+                "dimensionality": {
+                    "rows": len(data),
+                    "columns": list(data.columns),
+                    "memory_usage": f"{data.memory_usage().sum() / 1e6:.2f} MB"
+                },
+                "statistical_profile": data.describe(percentiles=[.25, .5, .75]).to_dict(),
+                "temporal_analysis": {
+                    "date_ranges": {
+                        col: {
+                            "min": data[col].min(),
+                            "max": data[col].max()
+                        } for col in data.select_dtypes(include='datetime').columns
+                    }
+                },
+                "data_quality": {
+                    "missing_values": data.isnull().sum().to_dict(),
+                    "duplicates": data.duplicated().sum(),
+                    "cardinality": {
+                        col: data[col].nunique() for col in data.columns
+                    }
+                }
+            }
+            return analysis
+        except Exception as e:
+            return {"error": f"EDA Failed: {str(e)}"}
+
+class DistributionVisualizer(DataAnalyzer):
+    """Distribution visualizations"""
+    def invoke(self, data: pd.DataFrame, columns: List[str], **kwargs) -> str:
+      try:
+          plt.figure(figsize=(12, 6))
+          for i, col in enumerate(columns, 1):
+              plt.subplot(1, len(columns), i)
+              sns.histplot(data[col], kde=True, stat="density")
+              plt.title(f'Distribution of {col}', fontsize=10)
+              plt.xticks(fontsize=8)
+              plt.yticks(fontsize=8)
+          plt.tight_layout()
+        
+          buf = io.BytesIO()
+          plt.savefig(buf, format='png', dpi=300, bbox_inches='tight')
+          plt.close()
+          return base64.b64encode(buf.getvalue()).decode()
+      except Exception as e:
+          return f"Visualization Error: {str(e)}"
+
+class TemporalAnalyzer(DataAnalyzer):
+    """Time series analysis"""
+    def invoke(self, data: pd.DataFrame, time_col: str, value_col: str, **kwargs) -> Dict[str, Any]:
+        try:
+            ts_data = data.set_index(pd.to_datetime(data[time_col]))[value_col]
+            decomposition = seasonal_decompose(ts_data, period=365)
+            
+            plt.figure(figsize=(12, 8))
+            decomposition.plot()
+            plt.tight_layout()
+            
+            buf = io.BytesIO()
+            plt.savefig(buf, format='png')
+            plt.close()
+            plot_data = base64.b64encode(buf.getvalue()).decode()
+            
+            return {
+                "trend_statistics": {
+                    "stationarity": adfuller(ts_data)[1],
+                    "seasonality_strength": max(decomposition.seasonal)
+                },
+                "visualization": plot_data
+            }
+        except Exception as e:
+            return {"error": f"Temporal Analysis Failed: {str(e)}"}
 
-class CSVDataSource(DataSource):
-    """Data source for CSV files."""
-    def __init__(self, file_path: str):
-        self.file_path = file_path
-        self.data: Optional[pd.DataFrame] = None
-    
-    def connect(self):
-        self.data = pd.read_csv(self.file_path)
-    
-    def fetch_data(self, query: str = None, **kwargs) -> pd.DataFrame:
-      if self.data is None:
-        raise Exception("No connection is made, call connect()")
-      return self.data
-    
-class DatabaseSource(DataSource):
-   def __init__(self, connection_string: str, database_type: str):
-        self.connection_string = connection_string
-        self.database_type = database_type
-        self.connection = None
-    
-   def connect(self):
-     if self.database_type.lower() == "sql":
-        #Placeholder for the actual database connection
-        self.connection = "Connected to SQL Database"
-     else:
-        raise Exception(f"Database type '{self.database_type}' is not supported")
+class HypothesisTester(DataAnalyzer):
+    """Statistical hypothesis testing"""
+    def invoke(self, data: pd.DataFrame, group_col: str, value_col: str, **kwargs) -> Dict[str, Any]:
+      try:
+        groups = data[group_col].unique()
       
-   def fetch_data(self, query: str, **kwargs) -> pd.DataFrame:
-        if self.connection is None:
-            raise Exception("No connection is made, call connect()")
-        #Placeholder for the data fetching
-        return pd.DataFrame({"result":[f"Fetched data based on query: {query}"]})
-     
+        if len(groups) < 2:
+            return {"error": "Insufficient groups for comparison"}
+        
+        if len(groups) == 2:
+            group_data = [data[data[group_col] == g][value_col] for g in groups]
+            stat, p = ttest_ind(*group_data)
+            test_type = "Independent t-test"
+        else:
+          group_data = [data[data[group_col] == g][value_col] for g in groups]
+          stat, p = f_oneway(*group_data)
+          test_type = "ANOVA"
+        
+        return {
+            "test_type": test_type,
+            "test_statistic": stat,
+            "p_value": p,
+            "effect_size": {
+              "cohens_d": abs(group_data[0].mean() - group_data[1].mean())/np.sqrt(
+                    (group_data[0].var() + group_data[1].var())/2
+              ) if len(groups) == 2 else None
+            },
+            "interpretation": self.interpret_p_value(p)
+        }
+      except Exception as e:
+        return {"error": f"Hypothesis Testing Failed: {str(e)}"}
+
+    def interpret_p_value(self, p: float) -> str:
+      if p < 0.001: return "Very strong evidence against H0"
+      elif p < 0.01: return "Strong evidence against H0"
+      elif p < 0.05: return "Evidence against H0"
+      elif p < 0.1: return "Weak evidence against H0"
+      else: return "No significant evidence against H0"
+
+class LogisticRegressionTrainer(DataAnalyzer):
+    """Logistic Regression Model Trainer"""
+    def invoke(self, data: pd.DataFrame, target_col: str, columns: List[str], **kwargs) -> Dict[str, Any]:
+      try:
+        X = data[columns]
+        y = data[target_col]
+        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
+        model = LogisticRegression(max_iter=1000)
+        model.fit(X_train, y_train)
+        y_pred = model.predict(X_test)
+        accuracy = accuracy_score(y_test, y_pred)
+        return {
+          "model_type": "Logistic Regression",
+           "accuracy": accuracy,
+           "model_params": model.get_params()
+         }
+      except Exception as e:
+         return {"error": f"Logistic Regression Model Error: {str(e)}"}
 
-class DataIngestion:
-    def __init__(self):
-      self.sources : Dict[str, DataSource] = {}
-    
-    def add_source(self, source_name: str, source: DataSource):
-         self.sources[source_name] = source
-    
-    def ingest_data(self, source_name: str, query: str = None, **kwargs) -> pd.DataFrame:
-       if source_name not in self.sources:
-         raise Exception(f"Source '{source_name}' not found")
-       source = self.sources[source_name]
-       source.connect()
-       return source.fetch_data(query, **kwargs)
-       
-class DataModel(BaseModel):
-  name : str
-  kpis : List[str] = Field(default_factory=list)
-  dimensions : List[str] = Field(default_factory=list)
-  custom_calculations : Optional[Dict[str, str]] = None
-  relations: Optional[Dict[str,str]] = None #Example {table1: table2}
-  
-  def to_json(self):
-    return json.dumps(self.dict())
-  
-  @staticmethod
-  def from_json(json_str):
-    return DataModel(**json.loads(json_str))
-  
-class DataModelling():
-    def __init__(self):
-      self.models : Dict[str, DataModel] = {}
-    
-    def add_model(self, model:DataModel):
-       self.models[model.name] = model
-    
-    def get_model(self, model_name: str) -> DataModel:
-      if model_name not in self.models:
-          raise Exception(f"Model '{model_name}' not found")
-      return self.models[model_name]
+# ---------------------- Groq Research Agent ---------------------------
+
+class GroqResearcher:
+    """Advanced AI Research Engine using Groq"""
+    def __init__(self, model_name="mixtral-8x7b-32768"):
+        self.model_name = model_name
+        self.system_template = """You are a senior data scientist at a research institution. 
+        Analyze this dataset with rigorous statistical methods and provide academic-quality insights:
+        {dataset_info}
+        
+        User Question: {query}
+        
+        Required Format:
+        - Executive Summary (1 paragraph)
+        - Methodology (bullet points)
+        - Key Findings (numbered list)
+        - Limitations
+        - Recommended Next Steps"""
+
+    def research(self, query: str, data: pd.DataFrame) -> str:
+        """Conduct academic-level analysis using Groq"""
+        try:
+            dataset_info = f"""
+            Dataset Dimensions: {data.shape}
+            Variables: {', '.join(data.columns)}
+            Temporal Coverage: {data.select_dtypes(include='datetime').columns.tolist()}
+            Missing Values: {data.isnull().sum().to_dict()}
+            """
+            
+            prompt = PromptTemplate.from_template(self.system_template).format(
+                dataset_info=dataset_info,
+                query=query
+            )
+            
+            completion = client.chat.completions.create(
+                messages=[
+                    {"role": "system", "content": "You are a research AI assistant"},
+                    {"role": "user", "content": prompt}
+                ],
+                model=self.model_name,
+                temperature=0.2,
+                max_tokens=4096,
+                stream=False
+            )
+            
+            return completion.choices[0].message.content
+        
+        except Exception as e:
+            return f"Research Error: {str(e)}"
 # ---------------------- Business Logic Layer ---------------------------
 class BusinessRule(BaseModel):
   name: str
@@ -228,56 +352,93 @@ class AutomatedReports():
          st.subheader(f"Data: {df_name}")
          st.table(df)
 
-# ---------------------- Groq Research Agent ---------------------------
+# ---------------------- Data Acquisition Layer ---------------------------
+class DataSource(ABC):
+    """Base class for data sources."""
+    @abstractmethod
+    def connect(self) -> None:
+        """Connect to the data source."""
+        pass
+    
+    @abstractmethod
+    def fetch_data(self, query: str, **kwargs) -> pd.DataFrame:
+         """Fetch the data based on a specific query."""
+         pass
 
-class GroqResearcher:
-    """Advanced AI Research Engine using Groq"""
-    def __init__(self, model_name="mixtral-8x7b-32768"):
-        self.model_name = model_name
-        self.system_template = """You are a senior data scientist at a research institution. 
-        Analyze this dataset with rigorous statistical methods and provide academic-quality insights:
-        {dataset_info}
-        
-        User Question: {query}
-        
-        Required Format:
-        - Executive Summary (1 paragraph)
-        - Methodology (bullet points)
-        - Key Findings (numbered list)
-        - Limitations
-        - Recommended Next Steps"""
 
-    def research(self, query: str, data: pd.DataFrame) -> str:
-        """Conduct academic-level analysis using Groq"""
-        try:
-            dataset_info = f"""
-            Dataset Dimensions: {data.shape}
-            Variables: {', '.join(data.columns)}
-            Temporal Coverage: {data.select_dtypes(include='datetime').columns.tolist()}
-            Missing Values: {data.isnull().sum().to_dict()}
-            """
-            
-            prompt = PromptTemplate.from_template(self.system_template).format(
-                dataset_info=dataset_info,
-                query=query
-            )
-            
-            completion = client.chat.completions.create(
-                messages=[
-                    {"role": "system", "content": "You are a research AI assistant"},
-                    {"role": "user", "content": prompt}
-                ],
-                model=self.model_name,
-                temperature=0.2,
-                max_tokens=4096,
-                stream=False
-            )
-            
-            return completion.choices[0].message.content
-        
-        except Exception as e:
-            return f"Research Error: {str(e)}"
+class CSVDataSource(DataSource):
+    """Data source for CSV files."""
+    def __init__(self, file_path: str):
+        self.file_path = file_path
+        self.data: Optional[pd.DataFrame] = None
+    
+    def connect(self):
+        self.data = pd.read_csv(self.file_path)
+    
+    def fetch_data(self, query: str = None, **kwargs) -> pd.DataFrame:
+      if self.data is None:
+        raise Exception("No connection is made, call connect()")
+      return self.data
+    
+class DatabaseSource(DataSource):
+   def __init__(self, connection_string: str, database_type: str):
+        self.connection_string = connection_string
+        self.database_type = database_type
+        self.connection = None
+    
+   def connect(self):
+     if self.database_type.lower() == "sql":
+        #Placeholder for the actual database connection
+        self.connection = "Connected to SQL Database"
+     else:
+        raise Exception(f"Database type '{self.database_type}' is not supported")
+      
+   def fetch_data(self, query: str, **kwargs) -> pd.DataFrame:
+        if self.connection is None:
+            raise Exception("No connection is made, call connect()")
+        #Placeholder for the data fetching
+        return pd.DataFrame({"result":[f"Fetched data based on query: {query}"]})
+     
 
+class DataIngestion:
+    def __init__(self):
+      self.sources : Dict[str, DataSource] = {}
+    
+    def add_source(self, source_name: str, source: DataSource):
+         self.sources[source_name] = source
+    
+    def ingest_data(self, source_name: str, query: str = None, **kwargs) -> pd.DataFrame:
+       if source_name not in self.sources:
+         raise Exception(f"Source '{source_name}' not found")
+       source = self.sources[source_name]
+       source.connect()
+       return source.fetch_data(query, **kwargs)
+       
+class DataModel(BaseModel):
+  name : str
+  kpis : List[str] = Field(default_factory=list)
+  dimensions : List[str] = Field(default_factory=list)
+  custom_calculations : Optional[Dict[str, str]] = None
+  relations: Optional[Dict[str,str]] = None #Example {table1: table2}
+  
+  def to_json(self):
+    return json.dumps(self.dict())
+  
+  @staticmethod
+  def from_json(json_str):
+    return DataModel(**json.loads(json_str))
+  
+class DataModelling():
+    def __init__(self):
+      self.models : Dict[str, DataModel] = {}
+    
+    def add_model(self, model:DataModel):
+       self.models[model.name] = model
+    
+    def get_model(self, model_name: str) -> DataModel:
+      if model_name not in self.models:
+          raise Exception(f"Model '{model_name}' not found")
+      return self.models[model_name]
 # ---------------------- Main Streamlit Application ---------------------------
 def main():
     st.set_page_config(page_title="AI BI Automation Platform", layout="wide")