Upload folder using huggingface_hub

reasoning/market_analysis.py

@@ -9,6 +9,8 @@ from datetime import datetime
 import numpy as np
 from collections import defaultdict
 
+from .base import ReasoningStrategy
+
 @dataclass
 class MarketSegment:
     """Market segment analysis."""
@@ -340,3 +342,109 @@ class MarketAnalyzer:
                 for t in self.trends
             ]
         }
+
+class MarketAnalysisStrategy(ReasoningStrategy):
+    """
+    Advanced market analysis strategy that combines multiple analytical tools
+    to provide comprehensive market insights.
+    """
+    
+    def __init__(self, config: Optional[Dict[str, Any]] = None):
+        """Initialize market analysis strategy."""
+        super().__init__()
+        self.config = config or {}
+        self.analyzer = MarketAnalyzer()
+    
+    async def reason(self, query: str, context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Perform market analysis based on query and context.
+        
+        Args:
+            query: The market analysis query
+            context: Additional context and parameters
+            
+        Returns:
+            Dict containing market analysis results and confidence scores
+        """
+        try:
+            # Extract market segment from query/context
+            segment = self._extract_segment(query, context)
+            
+            # Perform market analysis
+            analysis = await self._analyze_market(segment, context)
+            
+            # Get insights
+            insights = self.analyzer.get_market_insights()
+            
+            # Calculate confidence based on data quality and completeness
+            confidence = self._calculate_confidence(analysis, insights)
+            
+            return {
+                'answer': self._format_insights(insights),
+                'confidence': confidence,
+                'analysis': analysis,
+                'insights': insights,
+                'segment': segment
+            }
+            
+        except Exception as e:
+            logging.error(f"Market analysis failed: {str(e)}")
+            return {
+                'error': f"Market analysis failed: {str(e)}",
+                'confidence': 0.0
+            }
+    
+    def _extract_segment(self, query: str, context: Dict[str, Any]) -> str:
+        """Extract market segment from query and context."""
+        # Use context if available
+        if 'segment' in context:
+            return context['segment']
+            
+        # Default to general market
+        return 'general'
+    
+    async def _analyze_market(self, segment: str, context: Dict[str, Any]) -> Dict[str, Any]:
+        """Perform comprehensive market analysis."""
+        return await self.analyzer.analyze_market(segment, context)
+    
+    def _calculate_confidence(self, analysis: Dict[str, Any], insights: Dict[str, Any]) -> float:
+        """Calculate confidence score based on analysis quality."""
+        # Base confidence
+        confidence = 0.5
+        
+        # Adjust based on data completeness
+        if analysis.get('segment_analysis'):
+            confidence += 0.1
+        if analysis.get('competitor_analysis'):
+            confidence += 0.1
+        if analysis.get('trend_analysis'):
+            confidence += 0.1
+            
+        # Adjust based on insight quality
+        if insights.get('opportunities'):
+            confidence += 0.1
+        if insights.get('risks'):
+            confidence += 0.1
+            
+        return min(confidence, 1.0)
+    
+    def _format_insights(self, insights: Dict[str, Any]) -> str:
+        """Format market insights into readable text."""
+        sections = []
+        
+        if 'market_overview' in insights:
+            sections.append(f"Market Overview: {insights['market_overview']}")
+            
+        if 'opportunities' in insights:
+            opps = insights['opportunities']
+            sections.append("Key Opportunities:\n- " + "\n- ".join(opps))
+            
+        if 'risks' in insights:
+            risks = insights['risks']
+            sections.append("Key Risks:\n- " + "\n- ".join(risks))
+            
+        if 'recommendations' in insights:
+            recs = insights['recommendations']
+            sections.append("Recommendations:\n- " + "\n- ".join(recs))
+            
+        return "\n\n".join(sections)