Create components/statistical.py

components/statistical.py

@@ -0,0 +1,92 @@
+# components/statistical.py
+
+import numpy as np
+from scipy import stats
+from typing import Dict, List, Optional, Union
+import pandas as pd
+
+class StatisticalAnalyzer:
+    """Statistical analysis component"""
+    
+    @staticmethod
+    def analyze_distribution(data: Union[List[float], np.ndarray]) -> Dict:
+        """Analyze data distribution"""
+        result = {
+            "n_samples": len(data),
+            "mean": float(np.mean(data)),
+            "std": float(np.std(data)),
+            "median": float(np.median(data)),
+            "skewness": float(stats.skew(data)),
+            "kurtosis": float(stats.kurtosis(data))
+        }
+        
+        # Test for normality
+        statistic, p_value = stats.normaltest(data)
+        result["normality_test"] = {
+            "statistic": float(statistic),
+            "p_value": float(p_value),
+            "is_normal": p_value > 0.05
+        }
+        
+        return result
+    
+    @staticmethod
+    def calculate_confidence_interval(
+        data: Union[List[float], np.ndarray],
+        confidence: float = 0.95
+    ) -> Dict:
+        """Calculate confidence intervals"""
+        mean = np.mean(data)
+        std_err = stats.sem(data)
+        ci = stats.t.interval(confidence, len(data)-1, loc=mean, scale=std_err)
+        
+        return {
+            "mean": float(mean),
+            "ci_lower": float(ci[0]),
+            "ci_upper": float(ci[1]),
+            "confidence": confidence
+        }
+    
+    @staticmethod
+    def forecast_probability_cone(
+        data: Union[List[float], np.ndarray],
+        steps: int = 10,
+        confidence: float = 0.95
+    ) -> Dict:
+        """Generate probability cone forecast"""
+        mean = np.mean(data)
+        std_err = stats.sem(data)
+        t_value = stats.t.ppf((1 + confidence) / 2, len(data) - 1)
+        
+        time_points = list(range(steps))
+        means = [mean] * steps
+        errors = [t_value * std_err * np.sqrt(1 + i/len(data)) 
+                 for i in range(steps)]
+        
+        return {
+            "time": time_points,
+            "mean": means,
+            "lower": [m - e for m, e in zip(means, errors)],
+            "upper": [m + e for m, e in zip(means, errors)]
+        }
+        
+    @staticmethod
+    def analyze_correlations(df: pd.DataFrame) -> Dict:
+        """Analyze correlations between variables"""
+        corr_matrix = df.corr()
+        
+        # Find significant correlations
+        significant = []
+        for i in range(len(corr_matrix.columns)):
+            for j in range(i+1, len(corr_matrix.columns)):
+                if abs(corr_matrix.iloc[i,j]) > 0.5:
+                    significant.append({
+                        "var1": corr_matrix.columns[i],
+                        "var2": corr_matrix.columns[j],
+                        "correlation": float(corr_matrix.iloc[i,j])
+                    })
+        
+        return {
+            "correlation_matrix": corr_matrix.to_dict(),
+            "significant_correlations": significant
+        }