Spaces:
Sleeping
Sleeping
| """ | |
| Phase Scoring Utilities | |
| Provides utilities for calculating phase scores (0-10 scale) based on | |
| agent outputs and analysis results. | |
| Score interpretation: | |
| - 9-10: Strong bullish conviction | |
| - 7-8: Moderate bullish bias | |
| - 5-6: Neutral/mixed signals | |
| - 3-4: Moderate bearish bias | |
| - 0-2: Strong bearish conviction | |
| """ | |
| from typing import Any, Dict | |
| def calculate_technical_phase_score(state: Dict[str, Any]) -> float: | |
| """ | |
| Calculate score for technical analysis phase. | |
| Combines indicator, pattern, trend, and decision agent outputs. | |
| Args: | |
| state: Workflow state containing technical analysis results | |
| Returns: | |
| Score from 0-10 (0=bearish, 5=neutral, 10=bullish) | |
| """ | |
| scores = [] | |
| # Indicator analysis contribution | |
| indicator_analysis = state.get("indicator_analysis", {}) | |
| if indicator_analysis: | |
| rsi_value = indicator_analysis.get("rsi", {}).get("value") | |
| macd_histogram = indicator_analysis.get("macd", {}).get("histogram") | |
| stochastic = indicator_analysis.get("stochastic", {}) | |
| indicator_score = 5.0 # Start neutral | |
| # RSI contribution (-1 to +1) | |
| if rsi_value is not None: | |
| if rsi_value < 30: | |
| indicator_score += 1.5 # Oversold = bullish | |
| elif rsi_value > 70: | |
| indicator_score -= 1.5 # Overbought = bearish | |
| elif rsi_value < 40: | |
| indicator_score += 0.5 | |
| elif rsi_value > 60: | |
| indicator_score -= 0.5 | |
| # MACD contribution | |
| if macd_histogram is not None: | |
| if macd_histogram > 0: | |
| indicator_score += 1.0 | |
| else: | |
| indicator_score -= 1.0 | |
| # Stochastic contribution | |
| if stochastic: | |
| k_value = stochastic.get("k") | |
| if k_value is not None: | |
| if k_value < 20: | |
| indicator_score += 1.0 | |
| elif k_value > 80: | |
| indicator_score -= 1.0 | |
| scores.append(max(0, min(10, indicator_score))) | |
| # Pattern analysis contribution | |
| pattern_analysis = state.get("pattern_analysis", {}) | |
| if pattern_analysis: | |
| bullish_patterns = pattern_analysis.get("bullish_patterns", []) | |
| bearish_patterns = pattern_analysis.get("bearish_patterns", []) | |
| pattern_score = 5.0 | |
| pattern_score += len(bullish_patterns) * 0.5 | |
| pattern_score -= len(bearish_patterns) * 0.5 | |
| scores.append(max(0, min(10, pattern_score))) | |
| # Trend analysis contribution | |
| trend_analysis = state.get("trend_analysis", {}) | |
| if trend_analysis: | |
| trend_direction = trend_analysis.get("trend", {}).get("direction", "neutral") | |
| trend_strength = trend_analysis.get("trend", {}).get("strength", 0.5) | |
| if trend_direction == "bullish": | |
| trend_score = 5.0 + (trend_strength * 5.0) | |
| elif trend_direction == "bearish": | |
| trend_score = 5.0 - (trend_strength * 5.0) | |
| else: | |
| trend_score = 5.0 | |
| scores.append(max(0, min(10, trend_score))) | |
| # Decision analysis contribution | |
| decision_analysis = state.get("decision_analysis", {}) | |
| if decision_analysis: | |
| decision = decision_analysis.get("decision", "hold").lower() | |
| confidence = decision_analysis.get("confidence", 0.5) | |
| if decision == "buy": | |
| decision_score = 5.0 + (confidence * 5.0) | |
| elif decision == "sell": | |
| decision_score = 5.0 - (confidence * 5.0) | |
| else: | |
| decision_score = 5.0 | |
| scores.append(max(0, min(10, decision_score))) | |
| # Return weighted average (or neutral if no scores) | |
| if scores: | |
| return round(sum(scores) / len(scores), 1) | |
| return 5.0 | |
| def calculate_fundamental_phase_score(state: Dict[str, Any]) -> float: | |
| """ | |
| Calculate score for fundamental analysis phase. | |
| Combines fundamentals and sentiment agent outputs. | |
| Args: | |
| state: Workflow state containing fundamental analysis results | |
| Returns: | |
| Score from 0-10 (0=bearish, 5=neutral, 10=bullish) | |
| """ | |
| scores = [] | |
| # Fundamentals analysis contribution | |
| fundamentals_analysis = state.get("fundamentals_analysis", {}) | |
| if fundamentals_analysis: | |
| summary = fundamentals_analysis.get("summary", {}) | |
| financial_health = summary.get("financial_health", "moderate") | |
| valuation = summary.get("valuation", "fairly_valued") | |
| growth_potential = summary.get("growth_potential", "moderate") | |
| # Map qualitative assessments to scores | |
| health_map = {"strong": 8.5, "moderate": 5.0, "weak": 2.0} | |
| valuation_map = {"undervalued": 8.0, "fairly_valued": 5.0, "overvalued": 2.0} | |
| growth_map = {"high": 8.0, "moderate": 5.0, "low": 3.0} | |
| fundamentals_score = ( | |
| health_map.get(financial_health, 5.0) * 0.4 | |
| + valuation_map.get(valuation, 5.0) * 0.4 | |
| + growth_map.get(growth_potential, 5.0) * 0.2 | |
| ) | |
| scores.append(fundamentals_score) | |
| # Sentiment analysis contribution | |
| sentiment_analysis = state.get("sentiment_analysis", {}) | |
| if sentiment_analysis: | |
| sentiment_score_raw = sentiment_analysis.get("sentiment_score", 0.0) | |
| # Convert from -1/+1 scale to 0-10 scale | |
| sentiment_score = (sentiment_score_raw + 1) * 5.0 | |
| scores.append(sentiment_score) | |
| # Return average (or neutral if no scores) | |
| if scores: | |
| return round(sum(scores) / len(scores), 1) | |
| return 5.0 | |
| def calculate_sentiment_phase_score(state: Dict[str, Any]) -> float: | |
| """ | |
| Calculate score for sentiment analysis phase. | |
| Based on news agent outputs. | |
| Args: | |
| state: Workflow state containing sentiment analysis results | |
| Returns: | |
| Score from 0-10 (0=bearish, 5=neutral, 10=bullish) | |
| """ | |
| news_analysis = state.get("news_analysis", {}) | |
| if news_analysis: | |
| sentiment_score_raw = news_analysis.get("sentiment_score", 0.0) | |
| # Convert from -1/+1 scale to 0-10 scale | |
| sentiment_score = (sentiment_score_raw + 1) * 5.0 | |
| return round(sentiment_score, 1) | |
| return 5.0 | |
| def calculate_research_synthesis_phase_score(state: Dict[str, Any]) -> float: | |
| """ | |
| Calculate score for research synthesis phase. | |
| Combines technical analyst and researcher team outputs. | |
| Args: | |
| state: Workflow state containing research synthesis results | |
| Returns: | |
| Score from 0-10 (0=bearish, 5=neutral, 10=bullish) | |
| """ | |
| scores = [] | |
| # Technical analyst contribution | |
| technical_analyst = state.get("technical_analyst", {}) | |
| if technical_analyst: | |
| alignment = technical_analyst.get("alignment", {}) | |
| technical_bias = alignment.get("technical_bias", "neutral") | |
| alignment_score = alignment.get("alignment_score", 0.5) | |
| if technical_bias == "positive": | |
| analyst_score = 5.0 + (alignment_score * 5.0) | |
| elif technical_bias == "negative": | |
| analyst_score = 5.0 - (alignment_score * 5.0) | |
| else: | |
| analyst_score = 5.0 | |
| scores.append(analyst_score) | |
| # Researcher synthesis contribution | |
| researcher_synthesis = state.get("researcher_synthesis", {}) | |
| if researcher_synthesis: | |
| synthesis = researcher_synthesis.get("synthesis", {}) | |
| overall_lean = synthesis.get("overall_lean", "neutral") | |
| signal_ratio = synthesis.get("signal_ratio", 0.5) | |
| if overall_lean == "bullish": | |
| synthesis_score = 5.0 + (signal_ratio * 5.0) | |
| elif overall_lean == "bearish": | |
| synthesis_score = 5.0 - ((1 - signal_ratio) * 5.0) | |
| else: | |
| synthesis_score = 5.0 | |
| scores.append(synthesis_score) | |
| # Return average (or neutral if no scores) | |
| if scores: | |
| return round(sum(scores) / len(scores), 1) | |
| return 5.0 | |
| def calculate_risk_phase_score(state: Dict[str, Any]) -> float: | |
| """ | |
| Calculate score for risk assessment phase. | |
| Inverts risk score (low risk = high score). | |
| Args: | |
| state: Workflow state containing risk assessment results | |
| Returns: | |
| Score from 0-10 (0=high risk, 5=moderate risk, 10=low risk) | |
| """ | |
| risk_assessment = state.get("risk_assessment", {}) | |
| if risk_assessment: | |
| risk_score_raw = risk_assessment.get("risk_score", 50.0) # 0-100 scale | |
| # Invert: low risk (0) = high score (10), high risk (100) = low score (0) | |
| risk_score = 10 - (risk_score_raw / 10.0) | |
| return round(max(0, min(10, risk_score)), 1) | |
| return 5.0 | |
| def get_phase_weights(investment_style: str) -> Dict[str, float]: | |
| """ | |
| Get phase weights based on investment style. | |
| Args: | |
| investment_style: Investment style identifier | |
| Returns: | |
| Dictionary mapping phase names to weights (sum = 1.0) | |
| """ | |
| # Weight schemes per investment style | |
| if investment_style == "long_term": | |
| return { | |
| "fundamental": 0.40, | |
| "technical": 0.25, | |
| "sentiment": 0.20, | |
| "research_synthesis": 0.0, # Aggregates others | |
| "risk": 0.15, | |
| "decision": 0.0, # Aggregates others | |
| } | |
| elif investment_style == "swing_trading": | |
| return { | |
| "fundamental": 0.20, | |
| "technical": 0.40, | |
| "sentiment": 0.25, | |
| "research_synthesis": 0.0, # Aggregates others | |
| "risk": 0.15, | |
| "decision": 0.0, # Aggregates others | |
| } | |
| else: | |
| # Default balanced weights | |
| return { | |
| "fundamental": 0.30, | |
| "technical": 0.30, | |
| "sentiment": 0.20, | |
| "research_synthesis": 0.0, | |
| "risk": 0.20, | |
| "decision": 0.0, | |
| } | |
| def calculate_weighted_confidence( | |
| phase_scores: Dict[str, float], investment_style: str | |
| ) -> float: | |
| """ | |
| Calculate overall confidence percentage from phase scores. | |
| Args: | |
| phase_scores: Dictionary mapping phase names to scores (0-10) | |
| investment_style: Investment style identifier | |
| Returns: | |
| Confidence percentage (0-100) | |
| """ | |
| weights = get_phase_weights(investment_style) | |
| # Filter to only include phases that have scores and weights | |
| weighted_sum = 0.0 | |
| total_weight = 0.0 | |
| for phase, score in phase_scores.items(): | |
| weight = weights.get(phase, 0.0) | |
| if weight > 0 and score is not None: | |
| weighted_sum += score * weight | |
| total_weight += weight | |
| # Normalize if we don't have all phases | |
| if total_weight > 0: | |
| # Convert from 0-10 scale to 0-100 percentage | |
| confidence = (weighted_sum / total_weight) * 10.0 | |
| return round(confidence, 1) | |
| return 50.0 # Default neutral confidence | |