add valuation engine

.gitignore

@@ -33,4 +33,6 @@ src/core/trading/docs/
 trades.db
 *.db
 *.png
-*.jpg
+*.jpg
+#
+*.md

examples/valuation_example.py

@@ -0,0 +1,178 @@
+"""
+Example usage of the Valuation Engine
+
+This demonstrates how to use the comprehensive valuation engine to value stocks
+using multiple models (DCF, DDM, P/E Multiple, etc.)
+"""
+
+import asyncio
+from src.core.valuation_engine import ValuationEngine, ValuationAssumptions, DCFAssumptions
+
+
+async def basic_valuation_example():
+    """Basic valuation example with default assumptions"""
+    print("=" * 60)
+    print("BASIC VALUATION EXAMPLE")
+    print("=" * 60)
+
+    ticker = "AAPL"
+
+    async with ValuationEngine() as engine:
+        # Run comprehensive valuation
+        result = await engine.value_stock(ticker)
+
+        # Print results
+        print(f"\n{'='*60}")
+        print(f"VALUATION RESULTS FOR {ticker}")
+        print(f"{'='*60}")
+        print(f"\nCurrent Price: ${result.current_price:.2f}")
+
+        if result.weighted_fair_value:
+            print(f"Weighted Fair Value: ${result.weighted_fair_value:.2f}")
+            print(f"Upside/Downside: {result.upside_downside:+.1f}%")
+
+        print(f"\nFair Value Range:")
+        print(f"  Pessimistic (P10): ${result.percentile_10:.2f}")
+        print(f"  Base Case (P50):   ${result.percentile_50:.2f}")
+        print(f"  Optimistic (P90):  ${result.percentile_90:.2f}")
+
+        print(f"\nModels Used: {result.valid_models_count}")
+        print(f"Total Confidence: {result.total_confidence:.2f}")
+
+        print(f"\n{'='*60}")
+        print("INDIVIDUAL MODEL RESULTS")
+        print(f"{'='*60}")
+
+        for model, valuation_result in result.model_results.items():
+            if valuation_result.is_valid:
+                print(f"\n{model.value.upper()}:")
+                print(f"  Fair Value: ${valuation_result.fair_value_per_share:.2f}")
+                print(f"  Confidence: {valuation_result.confidence:.2f}")
+                if valuation_result.assumptions:
+                    print(f"  Assumptions: {valuation_result.assumptions}")
+            else:
+                print(f"\n{model.value.upper()}: FAILED")
+                print(f"  Error: {valuation_result.error}")
+
+
+async def custom_assumptions_example():
+    """Valuation with custom assumptions"""
+    print("\n\n" + "=" * 60)
+    print("CUSTOM ASSUMPTIONS EXAMPLE")
+    print("=" * 60)
+
+    ticker = "MSFT"
+
+    # Create custom assumptions
+    assumptions = ValuationAssumptions(
+        dcf=DCFAssumptions(
+            growth_rate=0.08,  # 8% growth
+            terminal_growth_rate=0.03,  # 3% terminal growth
+            projection_years=7
+        ),
+        risk_free_rate=0.045,  # 4.5% risk-free rate
+        market_return=0.11  # 11% expected market return
+    )
+
+    async with ValuationEngine() as engine:
+        result = await engine.value_stock(ticker, assumptions=assumptions)
+
+        print(f"\nValuation for {ticker} with custom assumptions:")
+        print(f"  Current Price: ${result.current_price:.2f}")
+        if result.weighted_fair_value:
+            print(f"  Fair Value: ${result.weighted_fair_value:.2f}")
+            print(f"  Upside/Downside: {result.upside_downside:+.1f}%")
+
+
+async def multiple_stocks_example():
+    """Value multiple stocks in parallel"""
+    print("\n\n" + "=" * 60)
+    print("MULTIPLE STOCKS EXAMPLE")
+    print("=" * 60)
+
+    tickers = ["AAPL", "MSFT", "JNJ"]
+
+    async with ValuationEngine() as engine:
+        # Run valuations in parallel
+        tasks = [engine.value_stock(ticker) for ticker in tickers]
+        results = await asyncio.gather(*tasks)
+
+        # Print summary
+        print(f"\n{'Ticker':<10} {'Current':<12} {'Fair Value':<12} {'Upside/Downside':<15} {'Models'}")
+        print("-" * 65)
+
+        for ticker, result in zip(tickers, results):
+            upside_str = f"{result.upside_downside:+.1f}%" if result.upside_downside else "N/A"
+            fair_value_str = f"${result.weighted_fair_value:.2f}" if result.weighted_fair_value else "N/A"
+
+            print(f"{ticker:<10} ${result.current_price:<11.2f} {fair_value_str:<12} {upside_str:<15} {result.valid_models_count}")
+
+
+async def quick_valuation_example():
+    """Quick valuation for fast lookup"""
+    print("\n\n" + "=" * 60)
+    print("QUICK VALUATION EXAMPLE")
+    print("=" * 60)
+
+    ticker = "GOOGL"
+
+    async with ValuationEngine() as engine:
+        # Get just the weighted fair value (faster)
+        fair_value = await engine.get_quick_valuation(ticker)
+
+        print(f"\nQuick valuation for {ticker}:")
+        if fair_value:
+            print(f"  Fair Value: ${fair_value:.2f}")
+        else:
+            print(f"  Could not calculate fair value")
+
+
+async def dividend_stock_example():
+    """Value a dividend-paying stock (DDM model will be used)"""
+    print("\n\n" + "=" * 60)
+    print("DIVIDEND STOCK EXAMPLE")
+    print("=" * 60)
+
+    ticker = "JNJ"  # Johnson & Johnson - dividend aristocrat
+
+    async with ValuationEngine() as engine:
+        result = await engine.value_stock(ticker)
+
+        print(f"\nValuation for dividend stock {ticker}:")
+        print(f"  Current Price: ${result.current_price:.2f}")
+
+        if result.weighted_fair_value:
+            print(f"  Fair Value: ${result.weighted_fair_value:.2f}")
+            print(f"  Upside/Downside: {result.upside_downside:+.1f}%")
+
+        # Check if DDM was used
+        from src.core.valuation_engine.core.models import ValuationModel
+        if ValuationModel.DDM in result.model_results:
+            ddm_result = result.model_results[ValuationModel.DDM]
+            if ddm_result.is_valid:
+                print(f"\n  DDM Model:")
+                print(f"    Fair Value: ${ddm_result.fair_value_per_share:.2f}")
+                print(f"    Confidence: {ddm_result.confidence:.2f}")
+                print(f"    Dividend Growth: {ddm_result.assumptions.get('dividend_growth_rate', 0)*100:.1f}%")
+
+
+async def main():
+    """Run all examples"""
+    print("\n" + "=" * 60)
+    print("VALUATION ENGINE - EXAMPLE USAGE")
+    print("=" * 60)
+
+    # Run examples
+    await basic_valuation_example()
+    await custom_assumptions_example()
+    await multiple_stocks_example()
+    await quick_valuation_example()
+    await dividend_stock_example()
+
+    print("\n" + "=" * 60)
+    print("EXAMPLES COMPLETE")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/core/ticker_scanner/growth_speed_analyzer.py

@@ -118,81 +118,27 @@ from sklearn.linear_model import LinearRegression
 
 from src.core.ticker_scanner.core_enums import GrowthCategory
 from src.core.ticker_scanner.growth_metrics import GrowthSpeedMetrics
+from src.core.ticker_scanner.timeframe_config import TIMEFRAME_CONFIG
 from src.telegram_bot.logger import main_logger as logger
 
 
-# Timeframe configuration with metric weights and thresholds for different analysis windows
-TIMEFRAME_CONFIG = {
-    "1d": {
-        "weights": {"cmgr": 0.25, "cagr": 0.0, "roc": 0.25, "momentum": 0.20, "r_squared": 0.15, "sharpe": 0.15},
-        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
-        "thresholds": {"velocity_cap": 100, "roc_threshold": 2.0},
-        "volatility_penalty": 3.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
-        "smoothness_bonus": 0.10,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
-        "description": "Intraday - High velocity, short-term momentum focus. CMGR emphasis for monthly growth context. Accepts up to 3% daily volatility"
-    },
-    "5d": {
-        "weights": {"cmgr": 0.25, "cagr": 0.0, "roc": 0.30, "momentum": 0.25, "r_squared": 0.15, "sharpe": 0.05},
-        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
-        "thresholds": {"velocity_cap": 90, "roc_threshold": 1.5},
-        "volatility_penalty": 2.5,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
-        "smoothness_bonus": 0.15,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
-        "description": "Weekly - CMGR and momentum emphasis with CAGR context. Accepts up to 2.5% daily volatility"
-    },
-    "1mo": {
-        "weights": {"cmgr": 0.35, "cagr": 0.0, "roc": 0.25, "momentum": 0.25, "r_squared": 0.10, "sharpe": 0.05},
-        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
-        "thresholds": {"velocity_cap": 85, "roc_threshold": 1.2},
-        "volatility_penalty": 2.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
-        "smoothness_bonus": 0.20,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
-        "description": "Monthly - CMGR primary metric with CAGR component. Accepts up to 2% daily volatility"
-    },
-    "3mo": {
-        "weights": {"cmgr": 0.40, "cagr": 0.0, "roc": 0.20, "momentum": 0.25, "r_squared": 0.10, "sharpe": 0.05},
-        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
-        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.8},
-        "volatility_penalty": 1.5,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
-        "smoothness_bonus": 0.20,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
-        "description": "Quarterly - CMGR and CAGR emphasis for sustainable growth. Accepts up to 1.5% daily volatility"
-    },
-    "6mo": {
-        "weights": {"cagr": 0.35, "r_squared": 0.20, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.10},
-        "indicator_bonuses": {"rsi": 0.03, "price_position": 0.02, "support_distance": 0.01},
-        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.5},
-        "volatility_penalty": 1.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
-        "smoothness_bonus": 0.0,  # No smoothness bonus for medium-term (already has high r_squared weight)
-        "description": "Half-yearly - CAGR and stability focus. Accepts up to 1% daily volatility"
-    },
-    "1y": {
-        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
-        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
-        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.3},
-        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
-        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
-        "description": "Yearly - Long-term stability and consistency. Growth stability emphasis"
-    },
-    "2y": {
-        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
-        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
-        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.25},
-        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
-        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
-        "description": "2-year - Long-term consistency"
-    },
-    "5y": {
-        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
-        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
-        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.2},
-        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
-        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
-        "description": "5-year - Long-term consistency"
-    },
-}
-
-
 class GrowthSpeedAnalyzer:
     """Advanced growth velocity and acceleration analysis with timeframe awareness"""
 
+    # Constants for metric calculations
+    SECONDS_PER_YEAR = 365.25 * 24 * 3600
+    TRADING_DAYS_PER_YEAR = 252
+    DAYS_PER_MONTH = 30.4375
+    SIX_MONTHS_TRADING_DAYS = 126
+    MAX_ANNUALIZATION_FACTOR = 252000
+    MIN_DATA_POINTS = 2
+    CMGR_NORMALIZATION_CAP = 25
+    CAGR_NORMALIZATION_CAP = 50
+    SHARPE_NORMALIZATION_CAP = 3.0
+    MOMENTUM_NORMALIZATION_RANGE = 100
+    DOWNTREND_PENALTY_MAX = 50
+    DOWNTREND_PENALTY_DIVISOR = 10
+
     @staticmethod
     def analyze(prices: np.ndarray, dates: pd.DatetimeIndex, timeframe: str = "1y") -> GrowthSpeedMetrics:
         """Comprehensive growth speed analysis with timeframe-aware metrics"""
@@ -203,8 +149,8 @@ class GrowthSpeedAnalyzer:
 
         # Allow fewer data points (e.g., for 1d or 5d analysis)
         # We need at least 2 points to calculate a slope/return
-        if len(prices) < 2:
-            raise ValueError(f"Insufficient data points: {len(prices)} < 2")
+        if len(prices) < GrowthSpeedAnalyzer.MIN_DATA_POINTS:
+            raise ValueError(f"Insufficient data points: {len(prices)} < {GrowthSpeedAnalyzer.MIN_DATA_POINTS}")
 
         # Ensure prices is 1D
         if prices.ndim > 1:
@@ -247,22 +193,9 @@ class GrowthSpeedAnalyzer:
             r_squared = 0
 
         # 2. Compound growth metrics
-        # Use total_seconds to handle intraday data (like 1d/5m) accurately
-        if len(dates) > 1:
-            years = (dates[-1] - dates[0]).total_seconds() / (365.25 * 24 * 3600)
-        else:
-            years = 0
-
+        years = GrowthSpeedAnalyzer._calculate_years_from_dates(dates)
         total_return = ((prices[-1] / prices[0]) - 1) * 100
-
-        # Avoid division by zero for very short timeframes
-        if years > 0:
-            try:
-                cagr = (pow(prices[-1] / prices[0], 1 / years) - 1) * 100
-            except:
-                cagr = 0
-        else:
-            cagr = 0
+        cagr = GrowthSpeedAnalyzer._calculate_cagr(prices, years)
 
         # Calculate rolling CAGR for recent growth context
         # For short timeframes (1d-3mo), use 30-day window; for others use 90-day window
@@ -274,33 +207,13 @@ class GrowthSpeedAnalyzer:
 
         # 3. Acceleration (second derivative)
         returns = np.diff(np.log(prices))
+        acceleration = 0
         if len(returns) > 1:
-            # Calculate annualization factor dynamically based on actual bar frequency
-            # Avoid hardcoding assumptions about candle intervals (5m, 1h, 1d, etc)
-            if len(dates) > 1:
-                # Calculate average seconds between bars
-                total_seconds = (dates[-1] - dates[0]).total_seconds()
-                avg_seconds_per_bar = total_seconds / (len(dates) - 1)
-
-                # For stocks/crypto trading: assume 252 trading days/year, 24/7 for crypto
-                # Use 365.25 * 24 * 3600 seconds per year for continuous markets
-                seconds_per_year = 365.25 * 24 * 3600
-                factor = seconds_per_year / avg_seconds_per_bar
-
-                # Cap factor to prevent extreme values from micro-intervals
-                # Maximum reasonable annualization is ~252000 (1 second bars)
-                factor = min(factor, 252000)
-            else:
-                factor = 252
-
+            factor = GrowthSpeedAnalyzer._calculate_annualization_factor(dates)
             acceleration = np.polyfit(range(len(returns)), returns, 1)[0] * factor
-        else:
-            acceleration = 0
 
         # 4. Recent momentum (last 6 months vs overall)
-        # fix for safe indexing
-        six_months_ago = max(0, n - 126)
-
+        six_months_ago = max(0, n - GrowthSpeedAnalyzer.SIX_MONTHS_TRADING_DAYS)
         recent_return = (prices[-1] / prices[six_months_ago] - 1) * 100 if six_months_ago < n else 0
         recent_momentum = recent_return - (total_return / years * 0.5) if years > 0 else 0
 
@@ -308,14 +221,15 @@ class GrowthSpeedAnalyzer:
         daily_returns = np.diff(prices) / prices[:-1]
 
         if len(daily_returns) > 0:
-            mean_return = np.mean(daily_returns) * 252  # Annualized
-            std_return = np.std(daily_returns) * np.sqrt(252)  # Annualized
+            mean_return = np.mean(daily_returns) * GrowthSpeedAnalyzer.TRADING_DAYS_PER_YEAR
+            std_return = np.std(daily_returns) * np.sqrt(GrowthSpeedAnalyzer.TRADING_DAYS_PER_YEAR)
 
             sharpe_ratio = mean_return / std_return if std_return > 0 else 0
 
             # Sortino ratio (downside deviation)
             downside_returns = daily_returns[daily_returns < 0]
-            downside_std = np.std(downside_returns) * np.sqrt(252) if len(downside_returns) > 0 else std_return
+            downside_std_factor = np.sqrt(GrowthSpeedAnalyzer.TRADING_DAYS_PER_YEAR)
+            downside_std = np.std(downside_returns) * downside_std_factor if len(downside_returns) > 0 else std_return
             sortino_ratio = mean_return / downside_std if downside_std > 0 else 0
         else:
             sharpe_ratio = sortino_ratio = 0
@@ -584,12 +498,12 @@ class GrowthSpeedAnalyzer:
         is_short_term = timeframe in ["1d", "5d", "1mo", "3mo"]
 
         # Normalize core metrics (shared between short and long-term)
-        cmgr_norm = GrowthSpeedAnalyzer._normalize_metric(cmgr, 25)  # Cap CMGR at 25% monthly
-        cagr_norm = GrowthSpeedAnalyzer._normalize_metric(cagr, 50)
+        cmgr_norm = GrowthSpeedAnalyzer._normalize_metric(cmgr, GrowthSpeedAnalyzer.CMGR_NORMALIZATION_CAP)
+        cagr_norm = GrowthSpeedAnalyzer._normalize_metric(cagr, GrowthSpeedAnalyzer.CAGR_NORMALIZATION_CAP)
         r_squared_norm = max(r_squared, 0)  # Already bounded [0,1]
         accel_norm = GrowthSpeedAnalyzer._normalize_metric(acceleration, 1, center=True)
-        sharpe_norm = GrowthSpeedAnalyzer._normalize_metric(sharpe, 3)
-        momentum_norm = GrowthSpeedAnalyzer._normalize_metric(momentum, 100, center=True)
+        sharpe_norm = GrowthSpeedAnalyzer._normalize_metric(sharpe, GrowthSpeedAnalyzer.SHARPE_NORMALIZATION_CAP)
+        momentum_norm = GrowthSpeedAnalyzer._normalize_metric(momentum, GrowthSpeedAnalyzer.MOMENTUM_NORMALIZATION_RANGE, center=True)
 
         if is_short_term:
             # For timeframes <= 3 months: use ROC and momentum emphasis
@@ -602,7 +516,10 @@ class GrowthSpeedAnalyzer:
             # Penalize downtrends: if price is below starting price, apply severe penalty
             if total_return < 0:
                 # Downtrend detected. Apply downtrend penalty proportional to magnitude
-                downtrend_penalty = min(abs(total_return) / 10, 50)  # Max 50% penalty for -100% loss
+                downtrend_penalty = min(
+                    abs(total_return) / GrowthSpeedAnalyzer.DOWNTREND_PENALTY_DIVISOR,
+                    GrowthSpeedAnalyzer.DOWNTREND_PENALTY_MAX
+                )
                 downtrend_factor = 1 - (downtrend_penalty / 100)
             else:
                 downtrend_factor = 1.0
@@ -685,6 +602,41 @@ class GrowthSpeedAnalyzer:
 
         return round(score, 2)
 
+    @staticmethod
+    def _calculate_years_from_dates(dates: pd.DatetimeIndex) -> float:
+        """Calculate elapsed years from date index."""
+        if len(dates) < 2:
+            return 0
+        return (dates[-1] - dates[0]).total_seconds() / GrowthSpeedAnalyzer.SECONDS_PER_YEAR
+
+    @staticmethod
+    def _calculate_annualization_factor(dates: pd.DatetimeIndex) -> float:
+        """Calculate dynamic annualization factor based on bar frequency."""
+        if len(dates) < 2:
+            return GrowthSpeedAnalyzer.TRADING_DAYS_PER_YEAR
+
+        total_seconds = (dates[-1] - dates[0]).total_seconds()
+        avg_seconds_per_bar = total_seconds / (len(dates) - 1)
+        factor = GrowthSpeedAnalyzer.SECONDS_PER_YEAR / avg_seconds_per_bar
+        return min(factor, GrowthSpeedAnalyzer.MAX_ANNUALIZATION_FACTOR)
+
+    @staticmethod
+    def _calculate_r_squared(prices: np.ndarray, predictions: np.ndarray) -> float:
+        """Calculate R-squared goodness of fit."""
+        ss_res = np.sum((prices - predictions) ** 2)
+        ss_tot = np.sum((prices - np.mean(prices)) ** 2)
+        return 1 - (ss_res / ss_tot) if ss_tot > 0 else 0
+
+    @staticmethod
+    def _calculate_cagr(prices: np.ndarray, years: float) -> float:
+        """Calculate Compound Annual Growth Rate (CAGR)."""
+        if years <= 0 or len(prices) < 2:
+            return 0
+        try:
+            return (pow(prices[-1] / prices[0], 1 / years) - 1) * 100
+        except (ValueError, ZeroDivisionError):
+            return 0
+
     @staticmethod
     def _categorize_growth(velocity_score: float) -> GrowthCategory:
         """Classify growth speed with dynamic thresholds"""

src/core/ticker_scanner/timeframe_config.py

@@ -0,0 +1,74 @@
+"""
+Timeframe-specific configuration for growth speed analysis.
+
+Defines metric weights, thresholds, and parameters for different analysis timeframes.
+Each timeframe has customized weights to emphasize different aspects of growth analysis.
+"""
+
+# Timeframe configuration with metric weights and thresholds for different analysis windows
+TIMEFRAME_CONFIG = {
+    "1d": {
+        "weights": {"cmgr": 0.25, "cagr": 0.0, "roc": 0.25, "momentum": 0.20, "r_squared": 0.15, "sharpe": 0.15},
+        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
+        "thresholds": {"velocity_cap": 100, "roc_threshold": 2.0},
+        "volatility_penalty": 3.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
+        "smoothness_bonus": 0.10,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
+        "description": "Intraday - High velocity, short-term momentum focus. CMGR emphasis for monthly growth context. Accepts up to 3% daily volatility"
+    },
+    "5d": {
+        "weights": {"cmgr": 0.25, "cagr": 0.0, "roc": 0.30, "momentum": 0.25, "r_squared": 0.15, "sharpe": 0.05},
+        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
+        "thresholds": {"velocity_cap": 90, "roc_threshold": 1.5},
+        "volatility_penalty": 2.5,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
+        "smoothness_bonus": 0.15,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
+        "description": "Weekly - CMGR and momentum emphasis with CAGR context. Accepts up to 2.5% daily volatility"
+    },
+    "1mo": {
+        "weights": {"cmgr": 0.35, "cagr": 0.0, "roc": 0.25, "momentum": 0.25, "r_squared": 0.10, "sharpe": 0.05},
+        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
+        "thresholds": {"velocity_cap": 85, "roc_threshold": 1.2},
+        "volatility_penalty": 2.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
+        "smoothness_bonus": 0.20,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
+        "description": "Monthly - CMGR primary metric with CAGR component. Accepts up to 2% daily volatility"
+    },
+    "3mo": {
+        "weights": {"cmgr": 0.40, "cagr": 0.0, "roc": 0.20, "momentum": 0.25, "r_squared": 0.10, "sharpe": 0.05},
+        "indicator_bonuses": {"rsi": 0.05, "price_position": 0.03, "support_distance": 0.02},
+        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.8},
+        "volatility_penalty": 1.5,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
+        "smoothness_bonus": 0.20,  # Bonus (0-1) for consistent growth: reward r_squared * bonus if vol < threshold
+        "description": "Quarterly - CMGR and CAGR emphasis for sustainable growth. Accepts up to 1.5% daily volatility"
+    },
+    "6mo": {
+        "weights": {"cagr": 0.35, "r_squared": 0.20, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.10},
+        "indicator_bonuses": {"rsi": 0.03, "price_position": 0.02, "support_distance": 0.01},
+        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.5},
+        "volatility_penalty": 1.0,  # Volatility threshold (% daily). Above = penalize. Set to 0 to disable.
+        "smoothness_bonus": 0.0,  # No smoothness bonus for medium-term (already has high r_squared weight)
+        "description": "Half-yearly - CAGR and stability focus. Accepts up to 1% daily volatility"
+    },
+    "1y": {
+        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
+        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
+        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.3},
+        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
+        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
+        "description": "Yearly - Long-term stability and consistency. Growth stability emphasis"
+    },
+    "2y": {
+        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
+        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
+        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.25},
+        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
+        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
+        "description": "2-year - Long-term consistency"
+    },
+    "5y": {
+        "weights": {"cagr": 0.40, "r_squared": 0.25, "acceleration": 0.20, "sharpe": 0.15, "momentum": 0.00},
+        "indicator_bonuses": {"rsi": 0.02, "price_position": 0.01, "support_distance": 0.00},
+        "thresholds": {"velocity_cap": 80, "roc_threshold": 0.2},
+        "volatility_penalty": 0,  # No penalty for long-term (Sharpe ratio already accounts for vol)
+        "smoothness_bonus": 0.0,  # No smoothness bonus for long-term (already has high r_squared weight)
+        "description": "5-year - Long-term consistency"
+    },
+}

src/core/valuation_engine/__init__.py

@@ -0,0 +1,34 @@
+"""
+Comprehensive Valuation Engine
+
+Provides multi-model stock valuation with:
+- 11+ valuation models (DCF, DDM, RIM, Multiples, etc.)
+- Multi-source data fetching (yfinance, FMP, SEC)
+- Hybrid model selection (rule-based + ML)
+- Risk analysis (WACC, sensitivity, Monte Carlo)
+- Visual reporting
+"""
+
+from src.core.valuation_engine.orchestrator import ValuationEngine
+from src.core.valuation_engine.core.models import (
+    ValuationModel,
+    ValuationResult,
+    AggregatedValuation,
+    ScenarioResult,
+)
+from src.core.valuation_engine.core.assumptions import (
+    ValuationAssumptions,
+    DCFAssumptions,
+    MultiplesAssumptions,
+)
+
+__all__ = [
+    "ValuationEngine",
+    "ValuationModel",
+    "ValuationResult",
+    "AggregatedValuation",
+    "ScenarioResult",
+    "ValuationAssumptions",
+    "DCFAssumptions",
+    "MultiplesAssumptions",
+]

src/core/valuation_engine/aggregation/__init__.py

@@ -0,0 +1,6 @@
+"""Aggregation layer for combining valuation results"""
+
+from src.core.valuation_engine.aggregation.ensemble import EnsembleAggregator
+from src.core.valuation_engine.aggregation.percentile_calc import PercentileCalculator
+
+__all__ = ["EnsembleAggregator", "PercentileCalculator"]

src/core/valuation_engine/aggregation/ensemble.py

@@ -0,0 +1,152 @@
+"""Ensemble weighting for combining multiple valuation models"""
+
+from typing import Dict
+from src.core.valuation_engine.core.models import ValuationModel, ValuationResult
+
+
+class EnsembleAggregator:
+    """
+    Combines multiple valuation model results using confidence-weighted averaging.
+    """
+
+    def __init__(self, min_models_required: int = 3):
+        """
+        Initialize ensemble aggregator.
+
+        Args:
+            min_models_required: Minimum number of valid models required
+        """
+        self.min_models_required = min_models_required
+
+    def aggregate(self, results: Dict[ValuationModel, ValuationResult]) -> Dict:
+        """
+        Aggregate multiple valuation results into a single weighted fair value.
+
+        Strategy:
+        1. Filter valid results only
+        2. Normalize confidence scores to sum to 1.0
+        3. Calculate weighted average
+        4. Handle edge cases
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Dictionary with:
+            - weighted_fair_value: Confidence-weighted average
+            - valid_models_count: Number of valid models
+            - total_confidence: Sum of all confidence scores
+            - model_weights: Normalized weights used
+        """
+        # Step 1: Filter valid results
+        valid_results = {
+            model: result
+            for model, result in results.items()
+            if result.is_valid
+        }
+
+        # Check minimum models requirement
+        if len(valid_results) < self.min_models_required:
+            return {
+                'weighted_fair_value': None,
+                'valid_models_count': len(valid_results),
+                'total_confidence': 0.0,
+                'model_weights': {},
+                'error': f"Only {len(valid_results)} valid models, minimum required is {self.min_models_required}"
+            }
+
+        # Step 2: Extract fair values and confidences
+        fair_values = []
+        confidences = []
+        model_list = []
+
+        for model, result in valid_results.items():
+            fair_values.append(result.fair_value_per_share)
+            confidences.append(result.confidence)
+            model_list.append(model)
+
+        # Step 3: Normalize confidence scores
+        total_confidence = sum(confidences)
+
+        if total_confidence == 0:
+            # Fallback: equal weights if all confidences are 0
+            normalized_weights = [1.0 / len(confidences)] * len(confidences)
+        else:
+            normalized_weights = [c / total_confidence for c in confidences]
+
+        # Step 4: Calculate weighted average
+        weighted_fair_value = sum(
+            fv * weight
+            for fv, weight in zip(fair_values, normalized_weights)
+        )
+
+        # Step 5: Create weights dictionary for transparency
+        model_weights = dict(zip(model_list, normalized_weights))
+
+        return {
+            'weighted_fair_value': weighted_fair_value,
+            'valid_models_count': len(valid_results),
+            'total_confidence': total_confidence,
+            'model_weights': model_weights,
+            'error': None
+        }
+
+    def get_value_range(self, results: Dict[ValuationModel, ValuationResult]) -> tuple:
+        """
+        Get range of fair values from all valid models.
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Tuple of (min_value, max_value) or (None, None) if no valid results
+        """
+        valid_values = [
+            result.fair_value_per_share
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_values:
+            return (None, None)
+
+        return (min(valid_values), max(valid_values))
+
+    def get_confidence_weighted_median(self,
+                                       results: Dict[ValuationModel, ValuationResult]) -> float:
+        """
+        Calculate confidence-weighted median (alternative to mean).
+
+        This is more robust to outliers than weighted mean.
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Weighted median fair value or None
+        """
+        # Filter valid results
+        valid_results = [
+            (result.fair_value_per_share, result.confidence)
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_results:
+            return None
+
+        # Sort by fair value
+        sorted_results = sorted(valid_results, key=lambda x: x[0])
+
+        # Calculate cumulative weights
+        total_weight = sum(conf for _, conf in sorted_results)
+        cumulative_weight = 0
+        target_weight = total_weight / 2
+
+        # Find weighted median
+        for value, confidence in sorted_results:
+            cumulative_weight += confidence
+            if cumulative_weight >= target_weight:
+                return value
+
+        return sorted_results[-1][0]  # Fallback to highest value

src/core/valuation_engine/aggregation/percentile_calc.py

@@ -0,0 +1,208 @@
+"""Percentile calculation for fair value ranges"""
+
+from typing import Dict, List
+import numpy as np
+
+from src.core.valuation_engine.core.models import ValuationModel, ValuationResult
+
+
+class PercentileCalculator:
+    """
+    Calculates percentiles (P10, P50, P90) from valuation model results.
+    Provides pessimistic, base, and optimistic fair value estimates.
+    """
+
+    def __init__(self, method: str = "linear"):
+        """
+        Initialize percentile calculator.
+
+        Args:
+            method: Numpy interpolation method ('linear', 'lower', 'higher', 'midpoint', 'nearest')
+        """
+        self.method = method
+
+    def calculate_percentiles(self,
+                             results: Dict[ValuationModel, ValuationResult]) -> Dict[str, float]:
+        """
+        Calculate P10, P50 (median), and P90 percentiles from all valid model results.
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Dictionary with percentile_10, percentile_50, percentile_90
+        """
+        # Extract valid fair values
+        valid_values = [
+            result.fair_value_per_share
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_values:
+            return {
+                'percentile_10': None,
+                'percentile_50': None,
+                'percentile_90': None
+            }
+
+        # Convert to numpy array for percentile calculation
+        values_array = np.array(valid_values)
+
+        # Calculate percentiles
+        p10 = np.percentile(values_array, 10, method=self.method)
+        p50 = np.percentile(values_array, 50, method=self.method)
+        p90 = np.percentile(values_array, 90, method=self.method)
+
+        return {
+            'percentile_10': float(p10),
+            'percentile_50': float(p50),
+            'percentile_90': float(p90)
+        }
+
+    def calculate_confidence_weighted_percentiles(self,
+                                                   results: Dict[ValuationModel, ValuationResult]) -> Dict[str, float]:
+        """
+        Calculate percentiles weighted by confidence scores.
+
+        This gives more weight to models with higher confidence in the distribution.
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Dictionary with weighted percentiles
+        """
+        # Extract valid results
+        valid_results = [
+            (result.fair_value_per_share, result.confidence)
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_results:
+            return {
+                'percentile_10': None,
+                'percentile_50': None,
+                'percentile_90': None
+            }
+
+        # Sort by fair value
+        sorted_results = sorted(valid_results, key=lambda x: x[0])
+
+        # Calculate weighted percentiles
+        p10 = self._weighted_percentile(sorted_results, 0.10)
+        p50 = self._weighted_percentile(sorted_results, 0.50)
+        p90 = self._weighted_percentile(sorted_results, 0.90)
+
+        return {
+            'percentile_10': p10,
+            'percentile_50': p50,
+            'percentile_90': p90
+        }
+
+    def _weighted_percentile(self,
+                            sorted_results: List[tuple],
+                            percentile: float) -> float:
+        """
+        Calculate weighted percentile from sorted (value, weight) pairs.
+
+        Args:
+            sorted_results: List of (value, weight) tuples sorted by value
+            percentile: Target percentile (0.0 to 1.0)
+
+        Returns:
+            Weighted percentile value
+        """
+        if not sorted_results:
+            return None
+
+        # Calculate total weight and target cumulative weight
+        total_weight = sum(weight for _, weight in sorted_results)
+        target_weight = total_weight * percentile
+
+        # Find value at target cumulative weight
+        cumulative_weight = 0
+        for i, (value, weight) in enumerate(sorted_results):
+            cumulative_weight += weight
+
+            if cumulative_weight >= target_weight:
+                # Interpolate if not first element
+                if i > 0 and cumulative_weight > target_weight:
+                    prev_value, prev_weight = sorted_results[i-1]
+                    # Linear interpolation
+                    prev_cum_weight = cumulative_weight - weight
+                    weight_diff = cumulative_weight - prev_cum_weight
+                    target_offset = target_weight - prev_cum_weight
+
+                    if weight_diff > 0:
+                        interpolation_factor = target_offset / weight_diff
+                        return prev_value + interpolation_factor * (value - prev_value)
+
+                return value
+
+        # Fallback to last value
+        return sorted_results[-1][0]
+
+    def get_percentile_range(self,
+                            results: Dict[ValuationModel, ValuationResult],
+                            lower: float = 0.10,
+                            upper: float = 0.90) -> tuple:
+        """
+        Get percentile range (e.g., P10 to P90).
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+            lower: Lower percentile (default 0.10)
+            upper: Upper percentile (default 0.90)
+
+        Returns:
+            Tuple of (lower_percentile_value, upper_percentile_value)
+        """
+        valid_values = [
+            result.fair_value_per_share
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_values:
+            return (None, None)
+
+        values_array = np.array(valid_values)
+
+        lower_value = np.percentile(values_array, lower * 100, method=self.method)
+        upper_value = np.percentile(values_array, upper * 100, method=self.method)
+
+        return (float(lower_value), float(upper_value))
+
+    def calculate_all_percentiles(self,
+                                  results: Dict[ValuationModel, ValuationResult]) -> Dict[str, float]:
+        """
+        Calculate full set of percentiles (P5, P10, P25, P50, P75, P90, P95).
+
+        Useful for comprehensive distribution analysis.
+
+        Args:
+            results: Dictionary mapping ValuationModel to ValuationResult
+
+        Returns:
+            Dictionary with all percentiles
+        """
+        valid_values = [
+            result.fair_value_per_share
+            for result in results.values()
+            if result.is_valid
+        ]
+
+        if not valid_values:
+            return {f'percentile_{p}': None for p in [5, 10, 25, 50, 75, 90, 95]}
+
+        values_array = np.array(valid_values)
+
+        percentiles = {}
+        for p in [5, 10, 25, 50, 75, 90, 95]:
+            percentiles[f'percentile_{p}'] = float(
+                np.percentile(values_array, p, method=self.method)
+            )
+
+        return percentiles

src/core/valuation_engine/core/__init__.py

@@ -0,0 +1,37 @@
+"""Core data models and entities for valuation engine"""
+
+from src.core.valuation_engine.core.models import (
+    ValuationModel,
+    DataSource,
+    ValuationResult,
+    ScenarioResult,
+    AggregatedValuation,
+    SensitivityAnalysis,
+    MonteCarloResult,
+)
+from src.core.valuation_engine.core.assumptions import (
+    DCFAssumptions,
+    MultiplesAssumptions,
+    DDMAssumptions,
+    RIMAssumptions,
+    ScenarioAssumptions,
+    ValuationAssumptions,
+)
+from src.core.valuation_engine.core.stock_entity import StockEntity
+
+__all__ = [
+    "ValuationModel",
+    "DataSource",
+    "ValuationResult",
+    "ScenarioResult",
+    "AggregatedValuation",
+    "SensitivityAnalysis",
+    "MonteCarloResult",
+    "DCFAssumptions",
+    "MultiplesAssumptions",
+    "DDMAssumptions",
+    "RIMAssumptions",
+    "ScenarioAssumptions",
+    "ValuationAssumptions",
+    "StockEntity",
+]

src/core/valuation_engine/core/assumptions.py

@@ -0,0 +1,106 @@
+"""Valuation assumptions and configuration dataclasses"""
+
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+
+@dataclass
+class DCFAssumptions:
+    """Assumptions for DCF models"""
+    growth_rate: float = 0.05           # 5% growth
+    terminal_growth_rate: float = 0.025 # 2.5% perpetual growth
+    discount_rate: Optional[float] = None  # If None, calculate WACC
+    projection_years: int = 5
+    use_wacc: bool = True  # Use WACC for discount rate if not specified
+
+    def validate(self) -> bool:
+        """Validate assumptions are reasonable"""
+        if self.growth_rate < -0.5 or self.growth_rate > 1.0:
+            return False
+        if self.terminal_growth_rate < 0 or self.terminal_growth_rate > 0.05:
+            return False
+        if self.discount_rate and (self.discount_rate < 0 or self.discount_rate > 0.5):
+            return False
+        if self.projection_years < 1 or self.projection_years > 20:
+            return False
+        return True
+
+
+@dataclass
+class MultiplesAssumptions:
+    """Assumptions for multiples-based valuation"""
+    peer_group: List[str] = field(default_factory=list)
+    use_industry_median: bool = True  # Use median vs mean
+    use_trailing: bool = True  # Use trailing vs forward multiples
+    exclude_outliers: bool = True  # Exclude values beyond 3 std devs
+    min_peer_count: int = 3  # Minimum peers required
+
+
+@dataclass
+class DDMAssumptions:
+    """Assumptions for Dividend Discount Model"""
+    dividend_growth_rate: Optional[float] = None  # If None, calculate from history
+    discount_rate: Optional[float] = None  # If None, use WACC or cost of equity
+    use_cost_of_equity: bool = True  # Use cost of equity for discount rate
+    projection_years: int = 5  # For multi-stage DDM
+
+
+@dataclass
+class RIMAssumptions:
+    """Assumptions for Residual Income Model"""
+    cost_of_equity: Optional[float] = None  # If None, calculate using CAPM
+    projection_years: int = 5
+    terminal_growth_rate: float = 0.025
+    use_capm: bool = True
+
+
+@dataclass
+class ScenarioAssumptions:
+    """Assumptions for scenario-based valuation"""
+    # Bear scenario (pessimistic)
+    bear_growth_rate: float = 0.02
+    bear_margin_adjustment: float = -0.05  # -5% margin
+    bear_discount_rate_adjustment: float = 0.02  # +2% discount rate
+
+    # Base scenario (expected)
+    base_growth_rate: float = 0.05
+    base_margin_adjustment: float = 0.0
+    base_discount_rate_adjustment: float = 0.0
+
+    # Bull scenario (optimistic)
+    bull_growth_rate: float = 0.10
+    bull_margin_adjustment: float = 0.05  # +5% margin
+    bull_discount_rate_adjustment: float = -0.01  # -1% discount rate
+
+
+@dataclass
+class ValuationAssumptions:
+    """Global valuation assumptions consolidating all model assumptions"""
+    # Model-specific assumptions
+    dcf: DCFAssumptions = field(default_factory=DCFAssumptions)
+    multiples: MultiplesAssumptions = field(default_factory=MultiplesAssumptions)
+    ddm: DDMAssumptions = field(default_factory=DDMAssumptions)
+    rim: RIMAssumptions = field(default_factory=RIMAssumptions)
+    scenario: ScenarioAssumptions = field(default_factory=ScenarioAssumptions)
+
+    # Global financial assumptions
+    risk_free_rate: float = 0.04  # 4% risk-free rate (10-year Treasury)
+    market_return: float = 0.10   # 10% expected market return
+    tax_rate: float = 0.21        # 21% corporate tax rate
+
+    @property
+    def market_risk_premium(self) -> float:
+        """Calculate market risk premium"""
+        return self.market_return - self.risk_free_rate
+
+    def validate(self) -> bool:
+        """Validate all assumptions"""
+        if not self.dcf.validate():
+            return False
+        if self.risk_free_rate < 0 or self.risk_free_rate > 0.15:
+            return False
+        if self.market_return < 0 or self.market_return > 0.30:
+            return False
+        if self.tax_rate < 0 or self.tax_rate > 1.0:
+            return False
+        return True

src/core/valuation_engine/core/models.py

@@ -0,0 +1,137 @@
+"""Core data models for valuation engine"""
+
+from dataclasses import dataclass, field
+from typing import Optional, Dict, List
+from enum import Enum
+
+
+class ValuationModel(Enum):
+    """Supported valuation models"""
+    DCF = "dcf"
+    REVERSE_DCF = "reverse_dcf"
+    DDM = "ddm"
+    RESIDUAL_INCOME = "residual_income"
+    ASSET_BASED = "asset_based"
+    PE_MULTIPLE = "pe_multiple"
+    EV_EBITDA = "ev_ebitda"
+    EV_SALES = "ev_sales"
+    PEG = "peg"
+    GRAHAM = "graham"
+    SCENARIO_BASED = "scenario_based"
+    MARKET_IMPLIED = "market_implied"
+
+
+class DataSource(Enum):
+    """Data source for financial data"""
+    YFINANCE = "yfinance"
+    FMP = "fmp"
+    SEC = "sec"
+    CACHED = "cached"
+
+
+@dataclass
+class ValuationResult:
+    """Result from a single valuation model"""
+    model: ValuationModel
+    fair_value_per_share: Optional[float] = None
+    intrinsic_value: Optional[float] = None
+    confidence: float = 0.0  # 0-1 scale
+    data_source: DataSource = DataSource.YFINANCE
+    assumptions: Dict = field(default_factory=dict)
+    metadata: Dict = field(default_factory=dict)
+    error: Optional[str] = None
+
+    @property
+    def is_valid(self) -> bool:
+        """Check if this result is valid and can be used"""
+        return self.fair_value_per_share is not None and self.error is None
+
+
+@dataclass
+class ScenarioResult:
+    """Results for bear/base/bull scenarios"""
+    bear: Optional[float] = None
+    base: Optional[float] = None
+    bull: Optional[float] = None
+
+    def to_dict(self) -> Dict[str, Optional[float]]:
+        """Convert to dictionary"""
+        return {
+            "bear": self.bear,
+            "base": self.base,
+            "bull": self.bull
+        }
+
+
+@dataclass
+class AggregatedValuation:
+    """Final aggregated valuation result"""
+    ticker: str
+    current_price: float
+
+    # Individual model results
+    model_results: Dict[ValuationModel, ValuationResult] = field(default_factory=dict)
+
+    # Aggregated values
+    weighted_fair_value: Optional[float] = None  # Single weighted average
+    percentile_10: Optional[float] = None        # P10 (pessimistic)
+    percentile_50: Optional[float] = None        # P50 (median)
+    percentile_90: Optional[float] = None        # P90 (optimistic)
+
+    # Scenario results
+    scenario_results: Optional[ScenarioResult] = None
+
+    # Metadata
+    upside_downside: Optional[float] = None  # % from current price to weighted fair value
+    valid_models_count: int = 0
+    total_confidence: float = 0.0
+
+    @property
+    def fair_value_range(self) -> tuple:
+        """Returns (min, max) fair value range"""
+        valid_values = [r.fair_value_per_share for r in self.model_results.values() if r.is_valid]
+        if not valid_values:
+            return (None, None)
+        return (min(valid_values), max(valid_values))
+
+    def calculate_upside_downside(self) -> None:
+        """Calculate upside/downside percentage"""
+        if self.weighted_fair_value and self.current_price and self.current_price > 0:
+            self.upside_downside = ((self.weighted_fair_value - self.current_price) / self.current_price) * 100
+
+
+@dataclass
+class SensitivityAnalysis:
+    """Sensitivity analysis results"""
+    base_case_value: float
+    parameter_ranges: Dict[str, List[float]]  # param_name -> [values]
+    value_matrix: Dict[str, List[float]]      # param_name -> [resulting values]
+
+    def get_parameter_impact(self, parameter: str) -> Optional[List[float]]:
+        """Get value impact for a specific parameter"""
+        return self.value_matrix.get(parameter)
+
+
+@dataclass
+class MonteCarloResult:
+    """Monte Carlo simulation results"""
+    simulations: int
+    mean_value: float
+    median_value: float
+    std_dev: float
+    percentile_5: float
+    percentile_25: float
+    percentile_50: float
+    percentile_75: float
+    percentile_95: float
+    all_values: List[float] = field(default_factory=list)
+
+    @property
+    def confidence_interval_90(self) -> tuple:
+        """Returns 90% confidence interval (P5 to P95)"""
+        return (self.percentile_5, self.percentile_95)
+
+    @property
+    def confidence_interval_50(self) -> tuple:
+        """Returns 50% confidence interval (P25 to P75)"""
+        return (self.percentile_25, self.percentile_75)

src/core/valuation_engine/core/stock_entity.py

@@ -0,0 +1,187 @@
+"""Stock entity consolidating data from multiple sources"""
+
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Any
+import pandas as pd
+
+from src.core.fundamental_analysis.core_models import FinancialMetrics, TickerData
+from src.core.valuation_engine.core.models import DataSource
+
+
+@dataclass
+class StockEntity:
+    """
+    Consolidated stock data from multiple sources.
+    Integrates with existing FinancialMetrics but adds additional fields needed for valuation.
+    """
+    # Basic info
+    ticker: str
+    price: float
+    shares_outstanding: Optional[float] = None
+    market_cap: Optional[float] = None
+
+    # Company info
+    sector: Optional[str] = None
+    industry: Optional[str] = None
+    company_name: Optional[str] = None
+
+    # Financial metrics (from existing calculator)
+    financial_metrics: Optional[FinancialMetrics] = None
+
+    # Cash flow data
+    free_cash_flow: Optional[float] = None
+    operating_cash_flow: Optional[float] = None
+    capex: Optional[float] = None
+
+    # Income statement
+    revenue: Optional[float] = None
+    ebitda: Optional[float] = None
+    net_income: Optional[float] = None
+    eps: Optional[float] = None
+    operating_margin: Optional[float] = None
+
+    # Balance sheet
+    total_assets: Optional[float] = None
+    total_liabilities: Optional[float] = None
+    total_debt: Optional[float] = None
+    total_equity: Optional[float] = None
+    book_value: Optional[float] = None
+    book_value_per_share: Optional[float] = None
+
+    # Dividend data
+    dividend_per_share: Optional[float] = None
+    dividend_yield: Optional[float] = None
+    dividend_growth_rate: Optional[float] = None
+
+    # Growth metrics
+    revenue_growth: Optional[float] = None
+    earnings_growth: Optional[float] = None
+
+    # Risk metrics
+    beta: Optional[float] = None
+
+    # Historical data (for growth calculations)
+    historical_fcf: list = field(default_factory=list)
+    historical_eps: list = field(default_factory=list)
+    historical_revenue: list = field(default_factory=list)
+    historical_dividends: list = field(default_factory=list)
+
+    # Raw data objects
+    ticker_data: Optional[TickerData] = None  # yfinance data
+    fmp_data: Optional[Dict[str, Any]] = None  # FMP API data
+    sec_data: Optional[Dict[str, Any]] = None  # SEC API data
+
+    # Metadata
+    data_source: DataSource = DataSource.YFINANCE
+    data_completeness_score: float = 0.0
+    last_updated: Optional[str] = None
+
+    def calculate_completeness_score(self) -> float:
+        """
+        Calculate data completeness score (0-1).
+        Used for confidence weighting in model selection.
+        """
+        required_fields = [
+            self.price, self.shares_outstanding, self.sector,
+            self.free_cash_flow, self.revenue, self.net_income,
+            self.total_assets, self.total_liabilities, self.beta
+        ]
+        optional_fields = [
+            self.dividend_per_share, self.operating_margin,
+            self.book_value, self.ebitda, self.eps
+        ]
+
+        required_count = sum(1 for field in required_fields if field is not None)
+        optional_count = sum(1 for field in optional_fields if field is not None)
+
+        # Required fields are weighted more heavily
+        required_weight = 0.7
+        optional_weight = 0.3
+
+        required_score = required_count / len(required_fields)
+        optional_score = optional_count / len(optional_fields)
+
+        self.data_completeness_score = (required_score * required_weight +
+                                        optional_score * optional_weight)
+        return self.data_completeness_score
+
+    @property
+    def has_fcf(self) -> bool:
+        """Check if FCF data is available and positive"""
+        return self.free_cash_flow is not None and self.free_cash_flow > 0
+
+    @property
+    def has_dividends(self) -> bool:
+        """Check if company pays dividends"""
+        return (self.dividend_per_share is not None and
+                self.dividend_per_share > 0)
+
+    @property
+    def has_positive_earnings(self) -> bool:
+        """Check if company has positive earnings"""
+        return self.net_income is not None and self.net_income > 0
+
+    @property
+    def has_book_value(self) -> bool:
+        """Check if book value data is available"""
+        return self.book_value is not None and self.book_value > 0
+
+    @property
+    def debt_to_equity(self) -> Optional[float]:
+        """Calculate debt-to-equity ratio"""
+        if self.total_debt and self.total_equity and self.total_equity > 0:
+            return self.total_debt / self.total_equity
+        return None
+
+    @property
+    def is_financial_company(self) -> bool:
+        """Check if company is in financial sector"""
+        if not self.sector:
+            return False
+        financial_sectors = ["Financial", "Bank", "Insurance", "REIT"]
+        return any(sector.lower() in self.sector.lower() for sector in financial_sectors)
+
+    @property
+    def is_tech_company(self) -> bool:
+        """Check if company is in technology sector"""
+        if not self.sector:
+            return False
+        return "tech" in self.sector.lower() or "software" in self.sector.lower()
+
+    def get_market_cap_category(self) -> str:
+        """Categorize market cap (small/mid/large)"""
+        if not self.market_cap:
+            return "unknown"
+
+        # Market cap in billions
+        market_cap_b = self.market_cap / 1_000_000_000
+
+        if market_cap_b < 2:
+            return "small"
+        elif market_cap_b < 10:
+            return "mid"
+        else:
+            return "large"
+
+    @classmethod
+    def from_ticker_data(cls, ticker_data: TickerData, price: float) -> "StockEntity":
+        """
+        Create StockEntity from yfinance TickerData.
+        This is a convenience method for the most common case.
+        """
+        info = ticker_data.info or {}
+
+        return cls(
+            ticker=ticker_data.ticker,
+            price=price,
+            shares_outstanding=info.get("sharesOutstanding"),
+            market_cap=info.get("marketCap"),
+            sector=info.get("sector"),
+            industry=info.get("industry"),
+            company_name=info.get("longName"),
+            beta=info.get("beta"),
+            dividend_per_share=info.get("dividendRate"),
+            dividend_yield=info.get("dividendYield"),
+            ticker_data=ticker_data,
+            data_source=DataSource.YFINANCE
+        )

src/core/valuation_engine/data/__init__.py

@@ -0,0 +1,5 @@
+"""Data fetching layer with multi-source support"""
+
+from src.core.valuation_engine.data.multi_source_fetcher import MultiSourceDataFetcher
+
+__all__ = ["MultiSourceDataFetcher"]

src/core/valuation_engine/data/multi_source_fetcher.py

@@ -0,0 +1,273 @@
+"""
+Multi-source data fetcher with fallback logic.
+Primary: yfinance → Fallback 1: FMP → Fallback 2: SEC
+"""
+
+import asyncio
+from typing import Optional
+from datetime import datetime
+
+from src.core.fundamental_analysis.async_data_fetcher import AsyncDataFetcher
+from src.core.fundamental_analysis.data_extractor import DataExtractor
+from src.core.fundamental_analysis.metrics_calculator import MetricsCalculator
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.core.models import DataSource
+
+
+class MultiSourceDataFetcher:
+    """
+    Fetches stock data from multiple sources with intelligent fallback.
+    Strategy: yfinance (primary) → FMP (fallback) → SEC (fallback)
+    """
+
+    def __init__(self,
+                 fmp_api_key: Optional[str] = None,
+                 sec_api_key: Optional[str] = None,
+                 max_workers: int = 5):
+        """
+        Initialize multi-source fetcher.
+
+        Args:
+            fmp_api_key: Financial Modeling Prep API key (optional)
+            fmp_api_key: Financial Modeling Prep API key (optional)
+            sec_api_key: SEC API key (optional)
+            max_workers: Max workers for async data fetcher
+        """
+        self.yf_fetcher = AsyncDataFetcher(max_workers=max_workers)
+        self.fmp_api_key = fmp_api_key
+        self.sec_api_key = sec_api_key
+
+        # TODO: Initialize FMP and SEC clients when implemented
+        # self.fmp_client = FMPClient(fmp_api_key) if fmp_api_key else None
+        # self.sec_client = SECClient(sec_api_key) if sec_api_key else None
+
+    async def __aenter__(self):
+        """Async context manager entry"""
+        await self.yf_fetcher.__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit"""
+        await self.yf_fetcher.__aexit__(exc_type, exc_val, exc_tb)
+
+    async def fetch_comprehensive_data(self, ticker: str) -> StockEntity:
+        """
+        Fetch comprehensive stock data with fallback logic.
+
+        Args:
+            ticker: Stock ticker symbol
+
+        Returns:
+            StockEntity with consolidated data from all sources
+        """
+        # Step 1: Try yfinance (primary source)
+        stock_entity = await self._fetch_from_yfinance(ticker)
+
+        # Step 2: If yfinance succeeded, enrich with additional calculations
+        if stock_entity and stock_entity.ticker_data:
+            stock_entity = await self._enrich_with_calculations(stock_entity)
+
+        # Step 3: Check data completeness
+        if stock_entity:
+            stock_entity.calculate_completeness_score()
+
+        # TODO: Step 4: If data incomplete, try FMP fallback
+        # if stock_entity.data_completeness_score < 0.7 and self.fmp_api_key:
+        #     stock_entity = await self._enrich_with_fmp(stock_entity)
+
+        # TODO: Step 5: If still incomplete, try SEC fallback
+        # if stock_entity.data_completeness_score < 0.6 and self.sec_api_key:
+        #     stock_entity = await self._enrich_with_sec(stock_entity)
+
+        # Step 6: Set timestamp
+        if stock_entity:
+            stock_entity.last_updated = datetime.now().isoformat()
+
+        return stock_entity
+
+    async def _fetch_from_yfinance(self, ticker: str) -> Optional[StockEntity]:
+        """
+        Fetch data from yfinance.
+
+        Args:
+            ticker: Stock ticker symbol
+
+        Returns:
+            StockEntity or None if fetch failed
+        """
+        try:
+            # Fetch ticker data using existing AsyncDataFetcher
+            ticker_data = await self.yf_fetcher.fetch_ticker_data(ticker)
+
+            if not ticker_data or not ticker_data.info:
+                return None
+
+            # Get current price
+            price = ticker_data.info.get("currentPrice") or ticker_data.info.get("regularMarketPrice")
+            if not price:
+                return None
+
+            # Create StockEntity from ticker data
+            stock_entity = StockEntity.from_ticker_data(ticker_data, price)
+
+            # Extract additional data using DataExtractor
+            extractor = DataExtractor(ticker_data)
+
+            # Cash flow data
+            stock_entity.operating_cash_flow = extractor.get_cashflow_item(
+                "Operating Cash Flow",
+                ["Total Cash From Operating Activities", "CashFlowFromContinuingOperatingActivities"]
+            )
+            stock_entity.capex = extractor.get_cashflow_item(
+                "Capital Expenditure",
+                ["Capital Expenditures", "CapitalExpenditures"]
+            )
+
+            # Calculate FCF
+            if stock_entity.operating_cash_flow and stock_entity.capex:
+                stock_entity.free_cash_flow = stock_entity.operating_cash_flow + stock_entity.capex  # capex is negative
+
+            # Income statement
+            stock_entity.revenue = extractor.get_financial_item(
+                "Total Revenue",
+                ["totalRevenue", "Revenue"]
+            )
+            stock_entity.ebitda = extractor.get_financial_item(
+                "EBITDA",
+                ["ebitda"]
+            )
+            stock_entity.net_income = extractor.get_financial_item(
+                "Net Income",
+                ["netIncome", "Net Income"]
+            )
+
+            # Balance sheet
+            stock_entity.total_assets = extractor.get_balance_sheet_item(
+                "Total Assets",
+                ["totalAssets"]
+            )
+            stock_entity.total_liabilities = extractor.get_balance_sheet_item(
+                "Total Liabilities Net Minority Interest",
+                ["Total Liab", "totalLiabilities"]
+            )
+            stock_entity.total_debt = extractor.get_balance_sheet_item(
+                "Total Debt",
+                ["totalDebt", "Long Term Debt"]
+            )
+            stock_entity.total_equity = extractor.get_balance_sheet_item(
+                "Total Equity Gross Minority Interest",
+                ["Stockholders Equity", "totalStockholdersEquity"]
+            )
+
+            # Calculate book value
+            if stock_entity.total_assets and stock_entity.total_liabilities:
+                stock_entity.book_value = stock_entity.total_assets - stock_entity.total_liabilities
+
+            if stock_entity.book_value and stock_entity.shares_outstanding:
+                stock_entity.book_value_per_share = stock_entity.book_value / stock_entity.shares_outstanding
+
+            # Get historical data for growth calculations
+            stock_entity.historical_fcf = extractor.get_cashflow_historical(
+                "Free Cash Flow", 5, ["FreeCashFlow"]
+            )
+            stock_entity.historical_revenue = extractor.get_historical_data(
+                "Total Revenue", 5, ["totalRevenue", "Revenue"]
+            )
+
+            # EPS from info
+            stock_entity.eps = ticker_data.info.get("trailingEps")
+
+            # Operating margin
+            if stock_entity.net_income and stock_entity.revenue and stock_entity.revenue > 0:
+                stock_entity.operating_margin = stock_entity.net_income / stock_entity.revenue
+
+            return stock_entity
+
+        except Exception as e:
+            print(f"Error fetching from yfinance for {ticker}: {e}")
+            return None
+
+    async def _enrich_with_calculations(self, stock_entity: StockEntity) -> StockEntity:
+        """
+        Enrich stock entity with calculated metrics using MetricsCalculator.
+
+        Args:
+            stock_entity: StockEntity to enrich
+
+        Returns:
+            Enriched StockEntity
+        """
+        try:
+            if not stock_entity.ticker_data:
+                return stock_entity
+
+            # Use existing MetricsCalculator to calculate financial metrics
+            calculator = MetricsCalculator(DataExtractor(stock_entity.ticker_data))
+            financial_metrics = await calculator.calculate_all_metrics()
+
+            # Store financial metrics
+            stock_entity.financial_metrics = financial_metrics
+
+            # Extract specific metrics we need for valuation
+            if financial_metrics.roe:
+                # ROE is already calculated, we can use it for RIM
+
+                pass
+
+            # Calculate growth rates if historical data available
+            stock_entity.revenue_growth = self._calculate_growth_rate(
+                stock_entity.historical_revenue
+            )
+
+            # Calculate dividend growth rate if historical dividends available
+            if len(stock_entity.historical_dividends) > 1:
+                stock_entity.dividend_growth_rate = self._calculate_growth_rate(
+                    stock_entity.historical_dividends
+                )
+
+            return stock_entity
+
+        except Exception as e:
+            print(f"Error enriching with calculations: {e}")
+            return stock_entity
+
+    def _calculate_growth_rate(self, historical_values: list) -> Optional[float]:
+        """
+        Calculate CAGR from historical values.
+
+        Args:
+            historical_values: List of historical values (most recent first)
+
+        Returns:
+            CAGR as decimal (e.g., 0.05 for 5%) or None
+        """
+        try:
+            # Filter out None values
+            values = [v for v in historical_values if v is not None and v > 0]
+
+            if len(values) < 2:
+                return None
+
+            # CAGR = (Ending Value / Beginning Value) ^ (1 / Number of Years) - 1
+            beginning_value = values[-1]  # Oldest value
+            ending_value = values[0]      # Most recent value
+            num_years = len(values) - 1
+
+            if beginning_value <= 0:
+                return None
+
+            cagr = (ending_value / beginning_value) ** (1 / num_years) - 1
+            return cagr
+
+        except Exception:
+            return None
+
+    # TODO: Implement FMP fallback
+    # async def _enrich_with_fmp(self, stock_entity: StockEntity) -> StockEntity:
+    #     """Enrich with FMP API data"""
+    #     pass
+
+    # TODO: Implement SEC fallback
+    # async def _enrich_with_sec(self, stock_entity: StockEntity) -> StockEntity:
+    #     """Enrich with SEC EDGAR data"""
+    #     pass

src/core/valuation_engine/models/__init__.py

@@ -0,0 +1,5 @@
+"""Valuation models"""
+
+from src.core.valuation_engine.models.base import BaseValuationModel
+
+__all__ = ["BaseValuationModel"]

src/core/valuation_engine/models/base.py

@@ -0,0 +1,169 @@
+"""Abstract base class for all valuation models"""
+
+from abc import ABC, abstractmethod
+from typing import List
+import asyncio
+
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.core.assumptions import ValuationAssumptions
+from src.core.valuation_engine.core.models import ValuationResult, ValuationModel
+
+
+class BaseValuationModel(ABC):
+    """
+    Abstract base class for all valuation models.
+    All concrete valuation models must implement this interface.
+    """
+
+    def __init__(self):
+        self.model_type: ValuationModel = None  # Must be set by subclasses
+
+    @abstractmethod
+    def is_applicable(self, stock: StockEntity) -> bool:
+        """
+        Check if this model can be applied to this stock.
+
+        Args:
+            stock: StockEntity with company data
+
+        Returns:
+            True if model can be applied, False otherwise
+        """
+        pass
+
+    @abstractmethod
+    async def calculate(self,
+                       stock: StockEntity,
+                       assumptions: ValuationAssumptions) -> ValuationResult:
+        """
+        Calculate fair value using this model.
+
+        Args:
+            stock: StockEntity with company data
+            assumptions: ValuationAssumptions with model parameters
+
+        Returns:
+            ValuationResult with fair value and metadata
+        """
+        pass
+
+    @abstractmethod
+    def get_required_data(self) -> List[str]:
+        """
+        Return list of required data fields for this model.
+
+        Returns:
+            List of field names required (e.g., ['free_cash_flow', 'shares_outstanding'])
+        """
+        pass
+
+    def validate_inputs(self, stock: StockEntity) -> bool:
+        """
+        Validate that required data is available.
+
+        Args:
+            stock: StockEntity to validate
+
+        Returns:
+            True if all required data is present, False otherwise
+        """
+        required = self.get_required_data()
+        for field in required:
+            if not hasattr(stock, field):
+                return False
+            value = getattr(stock, field)
+            if value is None:
+                return False
+            # Check for numeric fields
+            if isinstance(value, (int, float)) and value == 0:
+                # Some fields can be zero (e.g., dividends), but not critical ones
+                if field in ['shares_outstanding', 'price']:
+                    return False
+        return True
+
+    async def calculate_safe(self,
+                            stock: StockEntity,
+                            assumptions: ValuationAssumptions) -> ValuationResult:
+        """
+        Safe wrapper around calculate that handles exceptions.
+
+        Args:
+            stock: StockEntity with company data
+            assumptions: ValuationAssumptions
+
+        Returns:
+            ValuationResult (with error field populated if calculation failed)
+        """
+        try:
+            # Check if model is applicable
+            if not self.is_applicable(stock):
+                return ValuationResult(
+                    model=self.model_type,
+                    error="Model not applicable to this stock",
+                    confidence=0.0
+                )
+
+            # Validate inputs
+            if not self.validate_inputs(stock):
+                return ValuationResult(
+                    model=self.model_type,
+                    error=f"Missing required data: {self.get_required_data()}",
+                    confidence=0.0
+                )
+
+            # Run calculation
+            result = await self.calculate(stock, assumptions)
+            return result
+
+        except ValueError as e:
+            return ValuationResult(
+                model=self.model_type,
+                error=f"Invalid inputs: {str(e)}",
+                confidence=0.0
+            )
+        except ZeroDivisionError:
+            return ValuationResult(
+                model=self.model_type,
+                error="Division by zero in calculation",
+                confidence=0.0
+            )
+        except Exception as e:
+            return ValuationResult(
+                model=self.model_type,
+                error=f"Calculation failed: {str(e)}",
+                confidence=0.0
+            )
+
+    def _run_in_executor(self, func, *args):
+        """
+        Run a CPU-bound function in executor to avoid blocking.
+
+        Args:
+            func: Function to run
+            *args: Arguments to pass to function
+
+        Returns:
+            Coroutine that resolves to function result
+        """
+        loop = asyncio.get_event_loop()
+        return loop.run_in_executor(None, func, *args)
+
+    def _calculate_confidence(self,
+                             stock: StockEntity,
+                             base_confidence: float = 0.7) -> float:
+        """
+        Calculate confidence score for this valuation.
+        Can be overridden by subclasses.
+
+        Args:
+            stock: StockEntity
+            base_confidence: Starting confidence (0-1)
+
+        Returns:
+            Adjusted confidence score (0-1)
+        """
+        # Adjust based on data completeness
+        completeness_score = stock.calculate_completeness_score()
+        adjusted_confidence = base_confidence * (0.7 + 0.3 * completeness_score)
+
+        return max(0.0, min(1.0, adjusted_confidence))

src/core/valuation_engine/models/intrinsic/__init__.py

@@ -0,0 +1,6 @@
+"""Intrinsic valuation models"""
+
+from src.core.valuation_engine.models.intrinsic.dcf import DCFModel
+from src.core.valuation_engine.models.intrinsic.ddm import DDMModel
+
+__all__ = ["DCFModel", "DDMModel"]

src/core/valuation_engine/models/intrinsic/dcf.py

@@ -0,0 +1,260 @@
+"""Discounted Cash Flow (DCF) valuation model"""
+
+from typing import List
+import asyncio
+
+from src.core.valuation_engine.models.base import BaseValuationModel
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.core.assumptions import ValuationAssumptions
+from src.core.valuation_engine.core.models import ValuationResult, ValuationModel, DataSource
+
+
+class DCFModel(BaseValuationModel):
+    """
+    Discounted Cash Flow (DCF) valuation model.
+
+    Formula:
+    1. Project future FCFs for N years
+    2. Calculate terminal value
+    3. Discount all cash flows to present value
+    4. Divide by shares outstanding
+
+    Best for: Mature companies with predictable FCF
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.model_type = ValuationModel.DCF
+
+    def is_applicable(self, stock: StockEntity) -> bool:
+        """
+        DCF is applicable if:
+        - FCF is available and positive
+        - Shares outstanding is available
+        """
+        return stock.has_fcf and stock.shares_outstanding and stock.shares_outstanding > 0
+
+    def get_required_data(self) -> List[str]:
+        """Required data fields for DCF"""
+        return ['free_cash_flow', 'shares_outstanding']
+
+    async def calculate(self,
+                       stock: StockEntity,
+                       assumptions: ValuationAssumptions) -> ValuationResult:
+        """
+        Calculate fair value using DCF.
+
+        Args:
+            stock: StockEntity with company data
+            assumptions: ValuationAssumptions
+
+        Returns:
+            ValuationResult with fair value
+        """
+        dcf_assumptions = assumptions.dcf
+
+        # Get base FCF
+        base_fcf = stock.free_cash_flow
+        shares = stock.shares_outstanding
+
+        # Determine discount rate (use WACC if specified, or from assumptions)
+        discount_rate = dcf_assumptions.discount_rate
+        if discount_rate is None:
+            # TODO: Calculate WACC when risk module is implemented
+            # For now, use default 10%
+            discount_rate = 0.10
+
+        # Get growth parameters
+        growth_rate = dcf_assumptions.growth_rate
+        terminal_growth_rate = dcf_assumptions.terminal_growth_rate
+        projection_years = dcf_assumptions.projection_years
+
+        # Adjust growth rate based on historical data if available
+        if stock.historical_fcf and len([f for f in stock.historical_fcf if f]) >= 2:
+            historical_growth = self._calculate_historical_growth(stock.historical_fcf)
+            if historical_growth is not None:
+                # Blend historical and assumed growth (70% historical, 30% assumption)
+                growth_rate = 0.7 * historical_growth + 0.3 * growth_rate
+
+        # Run calculation in executor (CPU-bound)
+        loop = asyncio.get_event_loop()
+        result_data = await loop.run_in_executor(
+            None,
+            self._calculate_dcf_sync,
+            base_fcf,
+            shares,
+            growth_rate,
+            discount_rate,
+            terminal_growth_rate,
+            projection_years
+        )
+
+        # Calculate confidence based on data quality
+        confidence = self._calculate_confidence_dcf(stock, growth_rate)
+
+        return ValuationResult(
+            model=self.model_type,
+            fair_value_per_share=result_data['fair_value_per_share'],
+            intrinsic_value=result_data['intrinsic_value'],
+            confidence=confidence,
+            data_source=stock.data_source,
+            assumptions={
+                'growth_rate': growth_rate,
+                'terminal_growth_rate': terminal_growth_rate,
+                'discount_rate': discount_rate,
+                'projection_years': projection_years,
+                'base_fcf': base_fcf
+            },
+            metadata={
+                'projected_fcfs': result_data['projected_fcfs'],
+                'terminal_value': result_data['terminal_value'],
+                'pv_fcfs': result_data['pv_fcfs'],
+                'pv_terminal': result_data['pv_terminal']
+            }
+        )
+
+    def _calculate_dcf_sync(self,
+                           base_fcf: float,
+                           shares_outstanding: float,
+                           growth_rate: float,
+                           discount_rate: float,
+                           terminal_growth_rate: float,
+                           projection_years: int) -> dict:
+        """
+        Synchronous DCF calculation (CPU-bound).
+        Refactored from existing DCFCalculator.
+
+        Args:
+            base_fcf: Base free cash flow
+            shares_outstanding: Number of shares
+            growth_rate: Projected growth rate
+            discount_rate: Discount rate (WACC)
+            terminal_growth_rate: Perpetual growth rate
+            projection_years: Number of projection years
+
+        Returns:
+            Dictionary with calculation results
+        """
+        # Step 1: Project future FCFs
+        projected_fcfs = []
+        for year in range(1, projection_years + 1):
+            fcf = base_fcf * ((1 + growth_rate) ** year)
+            projected_fcfs.append(fcf)
+
+        # Step 2: Calculate present values of projected FCFs
+        pv_fcfs = []
+        for year, fcf in enumerate(projected_fcfs, start=1):
+            pv = fcf / ((1 + discount_rate) ** year)
+            pv_fcfs.append(pv)
+
+        # Step 3: Calculate terminal value
+        last_fcf = projected_fcfs[-1]
+
+        # Check if discount rate > terminal growth rate (must be true for perpetuity formula)
+        if discount_rate > terminal_growth_rate:
+            terminal_value = last_fcf * (1 + terminal_growth_rate) / (discount_rate - terminal_growth_rate)
+        else:
+            # Fallback: use simple multiple of last FCF if terminal growth is too high
+            terminal_value = last_fcf * 10  # 10x last year FCF
+
+        # Step 4: Discount terminal value to present
+        pv_terminal = terminal_value / ((1 + discount_rate) ** projection_years)
+
+        # Step 5: Sum all present values to get intrinsic value
+        intrinsic_value = sum(pv_fcfs) + pv_terminal
+
+        # Step 6: Calculate fair value per share
+        fair_value_per_share = intrinsic_value / shares_outstanding
+
+        return {
+            'fair_value_per_share': fair_value_per_share,
+            'intrinsic_value': intrinsic_value,
+            'projected_fcfs': projected_fcfs,
+            'pv_fcfs': pv_fcfs,
+            'terminal_value': terminal_value,
+            'pv_terminal': pv_terminal
+        }
+
+    def _calculate_historical_growth(self, historical_fcf: List[float]) -> float:
+        """Calculate historical FCF growth rate (CAGR)"""
+        # Filter out None and negative values
+        valid_fcf = [fcf for fcf in historical_fcf if fcf is not None and fcf > 0]
+
+        if len(valid_fcf) < 2:
+            return None
+
+        # CAGR = (Ending / Beginning) ^ (1 / years) - 1
+        try:
+            beginning = valid_fcf[-1]  # Oldest
+            ending = valid_fcf[0]      # Most recent
+            years = len(valid_fcf) - 1
+
+            cagr = (ending / beginning) ** (1 / years) - 1
+
+            # Cap at reasonable range (-50% to +100%)
+            return max(-0.5, min(1.0, cagr))
+        except:
+            return None
+
+    def _calculate_confidence_dcf(self, stock: StockEntity, growth_rate: float) -> float:
+        """
+        Calculate confidence score for DCF valuation.
+
+        Factors:
+        - Data completeness
+        - Historical FCF consistency
+        - Growth rate reasonableness
+        - Company size/maturity
+        """
+        base_confidence = 0.75  # DCF is generally reliable
+
+        # Adjust for data completeness
+        completeness_factor = stock.calculate_completeness_score()
+
+        # Adjust for FCF history consistency
+        history_factor = 1.0
+        if stock.historical_fcf:
+            valid_fcf = [f for f in stock.historical_fcf if f is not None and f > 0]
+            if len(valid_fcf) >= 3:
+                # Check variance in FCF growth
+                growth_rates = []
+                for i in range(len(valid_fcf) - 1):
+                    if valid_fcf[i+1] > 0:
+                        gr = (valid_fcf[i] - valid_fcf[i+1]) / valid_fcf[i+1]
+                        growth_rates.append(gr)
+
+                if growth_rates:
+                    # Calculate coefficient of variation
+                    import statistics
+                    std_dev = statistics.stdev(growth_rates) if len(growth_rates) > 1 else 0
+                    mean_gr = statistics.mean(growth_rates)
+                    cv = abs(std_dev / mean_gr) if mean_gr != 0 else 2.0
+
+                    # Lower CV = more consistent = higher confidence
+                    # CV < 0.5: very consistent (bonus +0.1)
+                    # CV > 1.5: very volatile (penalty -0.2)
+                    if cv < 0.5:
+                        history_factor = 1.1
+                    elif cv > 1.5:
+                        history_factor = 0.8
+
+        # Adjust for growth rate reasonableness
+        growth_factor = 1.0
+        if growth_rate < 0 or growth_rate > 0.3:
+            # Very high or negative growth rates reduce confidence
+            growth_factor = 0.9
+
+        # Adjust for company size (larger companies = more predictable)
+        size_factor = 1.0
+        if stock.market_cap:
+            market_cap_category = stock.get_market_cap_category()
+            if market_cap_category == "large":
+                size_factor = 1.05
+            elif market_cap_category == "small":
+                size_factor = 0.95
+
+        # Calculate final confidence
+        confidence = base_confidence * completeness_factor * history_factor * growth_factor * size_factor
+
+        # Ensure confidence is in [0, 1] range
+        return max(0.0, min(1.0, confidence))

src/core/valuation_engine/models/intrinsic/ddm.py

@@ -0,0 +1,237 @@
+"""Dividend Discount Model (DDM) - Gordon Growth Model"""
+
+from typing import List
+import asyncio
+
+from src.core.valuation_engine.models.base import BaseValuationModel
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.core.assumptions import ValuationAssumptions
+from src.core.valuation_engine.core.models import ValuationResult, ValuationModel
+
+
+class DDMModel(BaseValuationModel):
+    """
+    Dividend Discount Model (Gordon Growth Model).
+
+    Formula: Price = D₁ / (r - g)
+    Where:
+    - D₁ = Next year's dividend
+    - r = Required rate of return (cost of equity)
+    - g = Dividend growth rate
+
+    Best for: Dividend-paying companies with stable dividends
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.model_type = ValuationModel.DDM
+
+    def is_applicable(self, stock: StockEntity) -> bool:
+        """
+        DDM is applicable if:
+        - Company pays dividends
+        - Dividend per share is positive
+        """
+        return stock.has_dividends
+
+    def get_required_data(self) -> List[str]:
+        """Required data fields for DDM"""
+        return ['dividend_per_share']
+
+    async def calculate(self,
+                       stock: StockEntity,
+                       assumptions: ValuationAssumptions) -> ValuationResult:
+        """
+        Calculate fair value using DDM.
+
+        Args:
+            stock: StockEntity with company data
+            assumptions: ValuationAssumptions
+
+        Returns:
+            ValuationResult with fair value
+        """
+        ddm_assumptions = assumptions.ddm
+
+        # Get current dividend
+        current_dividend = stock.dividend_per_share
+
+        # Determine dividend growth rate
+        dividend_growth = ddm_assumptions.dividend_growth_rate
+        if dividend_growth is None:
+            # Calculate from historical data if available
+            if stock.historical_dividends and len([d for d in stock.historical_dividends if d]) >= 2:
+                dividend_growth = self._calculate_dividend_growth(stock.historical_dividends)
+            else:
+                # Default to 3% if no historical data
+                dividend_growth = 0.03
+
+        # Determine discount rate (cost of equity)
+        discount_rate = ddm_assumptions.discount_rate
+        if discount_rate is None:
+            if ddm_assumptions.use_cost_of_equity:
+                # Calculate cost of equity using CAPM
+                discount_rate = self._calculate_cost_of_equity(stock, assumptions)
+            else:
+                # Default to 10% if can't calculate
+                discount_rate = 0.10
+
+        # Validate that r > g (required for Gordon Growth Model)
+        if discount_rate <= dividend_growth:
+            return ValuationResult(
+                model=self.model_type,
+                error=f"Discount rate ({discount_rate:.2%}) must be > growth rate ({dividend_growth:.2%})",
+                confidence=0.0
+            )
+
+        # Run calculation
+        loop = asyncio.get_event_loop()
+        fair_value = await loop.run_in_executor(
+            None,
+            self._calculate_ddm_sync,
+            current_dividend,
+            dividend_growth,
+            discount_rate
+        )
+
+        # Calculate confidence
+        confidence = self._calculate_confidence_ddm(stock, dividend_growth)
+
+        return ValuationResult(
+            model=self.model_type,
+            fair_value_per_share=fair_value,
+            confidence=confidence,
+            data_source=stock.data_source,
+            assumptions={
+                'current_dividend': current_dividend,
+                'dividend_growth_rate': dividend_growth,
+                'discount_rate': discount_rate
+            },
+            metadata={
+                'next_year_dividend': current_dividend * (1 + dividend_growth)
+            }
+        )
+
+    def _calculate_ddm_sync(self,
+                           current_dividend: float,
+                           growth_rate: float,
+                           discount_rate: float) -> float:
+        """
+        Synchronous DDM calculation.
+
+        Gordon Growth Model: P = D₁ / (r - g)
+        Where D₁ = D₀ × (1 + g)
+
+        Args:
+            current_dividend: Current dividend per share (D₀)
+            growth_rate: Dividend growth rate (g)
+            discount_rate: Cost of equity (r)
+
+        Returns:
+            Fair value per share
+        """
+        # Calculate next year's dividend
+        next_dividend = current_dividend * (1 + growth_rate)
+
+        # Apply Gordon Growth Model
+        fair_value = next_dividend / (discount_rate - growth_rate)
+
+        return fair_value
+
+    def _calculate_dividend_growth(self, historical_dividends: List[float]) -> float:
+        """Calculate historical dividend growth rate (CAGR)"""
+        # Filter out None values
+        valid_dividends = [d for d in historical_dividends if d is not None and d > 0]
+
+        if len(valid_dividends) < 2:
+            return 0.03  # Default 3%
+
+        try:
+            beginning = valid_dividends[-1]  # Oldest
+            ending = valid_dividends[0]      # Most recent
+            years = len(valid_dividends) - 1
+
+            cagr = (ending / beginning) ** (1 / years) - 1
+
+            # Cap at reasonable range (0% to 15% for dividend growth)
+            return max(0.0, min(0.15, cagr))
+        except:
+            return 0.03
+
+    def _calculate_cost_of_equity(self,
+                                  stock: StockEntity,
+                                  assumptions: ValuationAssumptions) -> float:
+        """
+        Calculate cost of equity using CAPM.
+
+        CAPM: Re = Rf + β(Rm - Rf)
+
+        Args:
+            stock: StockEntity
+            assumptions: ValuationAssumptions
+
+        Returns:
+            Cost of equity
+        """
+        risk_free_rate = assumptions.risk_free_rate
+        market_return = assumptions.market_return
+        beta = stock.beta if stock.beta else 1.0  # Default beta = 1
+
+        # CAPM formula
+        cost_of_equity = risk_free_rate + beta * (market_return - risk_free_rate)
+
+        # Ensure reasonable range (4% to 20%)
+        return max(0.04, min(0.20, cost_of_equity))
+
+    def _calculate_confidence_ddm(self, stock: StockEntity, growth_rate: float) -> float:
+        """
+        Calculate confidence score for DDM valuation.
+
+        Factors:
+        - Dividend history consistency
+        - Dividend payout ratio reasonableness
+        - Company maturity
+        - Growth rate reasonableness
+        """
+        base_confidence = 0.70  # DDM is reliable for dividend stocks
+
+        # Adjust for data completeness
+        completeness_factor = stock.calculate_completeness_score()
+
+        # Adjust for dividend history
+        history_factor = 1.0
+        if stock.historical_dividends:
+            valid_dividends = [d for d in stock.historical_dividends if d is not None and d > 0]
+            if len(valid_dividends) >= 3:
+                # Check if dividends are consistent (no cuts)
+                dividend_cuts = sum(1 for i in range(len(valid_dividends) - 1)
+                                  if valid_dividends[i] < valid_dividends[i + 1])
+                if dividend_cuts == 0:
+                    # No dividend cuts = dividend aristocrat = higher confidence
+                    history_factor = 1.15
+                elif dividend_cuts > 1:
+                    # Multiple cuts = less reliable
+                    history_factor = 0.85
+
+        # Adjust for growth rate reasonableness
+        growth_factor = 1.0
+        if growth_rate > 0.10:
+            # Very high dividend growth is unsustainable
+            growth_factor = 0.90
+        elif growth_rate < 0:
+            # Negative growth (dividend cuts expected)
+            growth_factor = 0.80
+
+        # Adjust for company size (larger = more stable dividends)
+        size_factor = 1.0
+        if stock.market_cap:
+            market_cap_category = stock.get_market_cap_category()
+            if market_cap_category == "large":
+                size_factor = 1.05
+            elif market_cap_category == "small":
+                size_factor = 0.95
+
+        # Calculate final confidence
+        confidence = base_confidence * completeness_factor * history_factor * growth_factor * size_factor
+
+        return max(0.0, min(1.0, confidence))

src/core/valuation_engine/models/relative/__init__.py

@@ -0,0 +1,5 @@
+"""Relative valuation models"""
+
+from src.core.valuation_engine.models.relative.multiples import PEMultipleModel
+
+__all__ = ["PEMultipleModel"]

src/core/valuation_engine/models/relative/multiples.py

@@ -0,0 +1,222 @@
+"""Price-to-Earnings (P/E) Multiple valuation model"""
+
+from typing import List, Optional
+import asyncio
+
+from src.core.valuation_engine.models.base import BaseValuationModel
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.core.assumptions import ValuationAssumptions
+from src.core.valuation_engine.core.models import ValuationResult, ValuationModel
+
+
+class PEMultipleModel(BaseValuationModel):
+    """
+    Price-to-Earnings (P/E) Multiple valuation model.
+
+    Strategy:
+    1. Get company's current P/E ratio
+    2. Compare to industry/sector median P/E
+    3. Calculate fair value: Fair P/E × EPS
+
+    Variations:
+    - Use peer group average
+    - Use historical average
+    - Use forward P/E vs trailing P/E
+
+    Best for: Quick relative valuation across peers
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.model_type = ValuationModel.PE_MULTIPLE
+
+    def is_applicable(self, stock: StockEntity) -> bool:
+        """
+        P/E Multiple is applicable if:
+        - Company has positive earnings (EPS > 0)
+        - P/E ratio is available
+        """
+        return (stock.has_positive_earnings and
+                stock.eps is not None and
+                stock.eps > 0 and
+                stock.financial_metrics is not None and
+                stock.financial_metrics.pe_ratio is not None)
+
+    def get_required_data(self) -> List[str]:
+        """Required data fields for P/E Multiple"""
+        return ['eps', 'price']
+
+    async def calculate(self,
+                       stock: StockEntity,
+                       assumptions: ValuationAssumptions) -> ValuationResult:
+        """
+        Calculate fair value using P/E Multiple.
+
+        Args:
+            stock: StockEntity with company data
+            assumptions: ValuationAssumptions
+
+        Returns:
+            ValuationResult with fair value
+        """
+        multiples_assumptions = assumptions.multiples
+
+        # Get current P/E ratio and EPS
+        current_pe = stock.financial_metrics.pe_ratio
+        eps = stock.eps
+
+        # Determine fair P/E (benchmark)
+        fair_pe = await self._get_fair_pe(stock, multiples_assumptions)
+
+        if fair_pe is None or fair_pe <= 0:
+            return ValuationResult(
+                model=self.model_type,
+                error="Could not determine fair P/E ratio",
+                confidence=0.0
+            )
+
+        # Calculate fair value
+        loop = asyncio.get_event_loop()
+        fair_value = await loop.run_in_executor(
+            None,
+            self._calculate_pe_valuation,
+            eps,
+            fair_pe
+        )
+
+        # Calculate confidence
+        confidence = self._calculate_confidence_pe(stock, current_pe, fair_pe)
+
+        return ValuationResult(
+            model=self.model_type,
+            fair_value_per_share=fair_value,
+            confidence=confidence,
+            data_source=stock.data_source,
+            assumptions={
+                'current_pe': current_pe,
+                'fair_pe': fair_pe,
+                'eps': eps
+            },
+            metadata={
+                'pe_premium_discount': ((current_pe - fair_pe) / fair_pe * 100) if fair_pe else None,
+                'sector': stock.sector
+            }
+        )
+
+    async def _get_fair_pe(self,
+                          stock: StockEntity,
+                          multiples_assumptions) -> Optional[float]:
+        """
+        Determine fair P/E ratio.
+
+        Strategy (in order of preference):
+        1. Use peer group average if specified
+        2. Use industry benchmark
+        3. Use historical company average
+        4. Use market average
+
+        Args:
+            stock: StockEntity
+            multiples_assumptions: MultiplesAssumptions
+
+        Returns:
+            Fair P/E ratio or None
+        """
+        # TODO: Implement peer group comparison when data available
+        # For now, use industry benchmarks based on sector
+
+        # Industry P/E benchmarks (approximate averages)
+        industry_pe_benchmarks = {
+            'Technology': 28.0,
+            'Healthcare': 25.0,
+            'Financial': 12.0,
+            'Financial Services': 12.0,
+            'Consumer Cyclical': 20.0,
+            'Consumer Defensive': 22.0,
+            'Energy': 15.0,
+            'Industrials': 20.0,
+            'Basic Materials': 18.0,
+            'Real Estate': 20.0,
+            'Utilities': 18.0,
+            'Communication Services': 22.0,
+            'Communication': 22.0,
+        }
+
+        # Try to get sector benchmark
+        if stock.sector and stock.sector in industry_pe_benchmarks:
+            return industry_pe_benchmarks[stock.sector]
+
+        # Fallback: use market average
+        return 20.0  # S&P 500 approximate average
+
+    def _calculate_pe_valuation(self, eps: float, fair_pe: float) -> float:
+        """
+        Calculate fair value using P/E multiple.
+
+        Formula: Fair Value = EPS × Fair P/E
+
+        Args:
+            eps: Earnings per share
+            fair_pe: Fair P/E ratio
+
+        Returns:
+            Fair value per share
+        """
+        return eps * fair_pe
+
+    def _calculate_confidence_pe(self,
+                                 stock: StockEntity,
+                                 current_pe: float,
+                                 fair_pe: float) -> float:
+        """
+        Calculate confidence score for P/E valuation.
+
+        Factors:
+        - Data quality
+        - How far current P/E is from fair P/E
+        - Earnings quality
+        - Company size
+        """
+        base_confidence = 0.65  # P/E is a quick relative measure, less precise than DCF
+
+        # Adjust for data completeness
+        completeness_factor = stock.calculate_completeness_score()
+
+        # Adjust for P/E deviation
+        deviation_factor = 1.0
+        if current_pe and fair_pe:
+            pe_ratio = current_pe / fair_pe
+            # If current P/E is very different from fair P/E, lower confidence
+            # (might indicate special circumstances not captured by simple P/E)
+            if pe_ratio < 0.5 or pe_ratio > 2.0:
+                # Very different from peer average
+                deviation_factor = 0.85
+            elif 0.7 < pe_ratio < 1.3:
+                # Close to peer average = higher confidence
+                deviation_factor = 1.1
+
+        # Adjust for earnings quality
+        quality_factor = 1.0
+        if stock.financial_metrics:
+            # Check if company has quality metrics
+            if stock.financial_metrics.quality_of_earnings:
+                # Quality of earnings > 1.0 is good (cash earnings > reported earnings)
+                qoe = stock.financial_metrics.quality_of_earnings
+                if qoe > 1.0:
+                    quality_factor = 1.05
+                elif qoe < 0.7:
+                    quality_factor = 0.90
+
+        # Adjust for company size (larger = more comparable to benchmarks)
+        size_factor = 1.0
+        if stock.market_cap:
+            market_cap_category = stock.get_market_cap_category()
+            if market_cap_category == "large":
+                size_factor = 1.05  # Large caps more comparable to industry averages
+            elif market_cap_category == "small":
+                size_factor = 0.95
+
+        # Calculate final confidence
+        confidence = base_confidence * completeness_factor * deviation_factor * quality_factor * size_factor
+
+        return max(0.0, min(1.0, confidence))

src/core/valuation_engine/orchestrator.py

@@ -0,0 +1,245 @@
+"""Main ValuationEngine orchestrator"""
+
+import asyncio
+from typing import Optional, Dict
+from datetime import datetime
+
+from src.core.valuation_engine.data.multi_source_fetcher import MultiSourceDataFetcher
+from src.core.valuation_engine.selector.model_selector import ModelSelector
+from src.core.valuation_engine.aggregation.ensemble import EnsembleAggregator
+from src.core.valuation_engine.aggregation.percentile_calc import PercentileCalculator
+from src.core.valuation_engine.core.models import (
+    ValuationModel,
+    ValuationResult,
+    AggregatedValuation
+)
+from src.core.valuation_engine.core.assumptions import ValuationAssumptions
+from src.core.valuation_engine.core.stock_entity import StockEntity
+
+# Import implemented models
+from src.core.valuation_engine.models.intrinsic.dcf import DCFModel
+from src.core.valuation_engine.models.intrinsic.ddm import DDMModel
+from src.core.valuation_engine.models.relative.multiples import PEMultipleModel
+
+
+class ValuationEngine:
+    """
+    Main orchestrator for comprehensive stock valuation.
+
+    Coordinates:
+    1. Data fetching (multi-source)
+    2. Model selection
+    3. Parallel model execution
+    4. Result aggregation
+    5. Risk analysis (TODO)
+    6. Report generation (TODO)
+    """
+
+    def __init__(self,
+                 fmp_api_key: Optional[str] = None,
+                 sec_api_key: Optional[str] = None,
+                 max_workers: int = 5,
+                 min_models_required: int = 2):
+        """
+        Initialize ValuationEngine.
+
+        Args:
+            fmp_api_key: Financial Modeling Prep API key (optional)
+            sec_api_key: SEC API key (optional)
+            max_workers: Max workers for async operations
+            min_models_required: Minimum valid models required for aggregation
+        """
+        self.data_fetcher = MultiSourceDataFetcher(
+            fmp_api_key=fmp_api_key,
+            sec_api_key=sec_api_key,
+            max_workers=max_workers
+        )
+        self.model_selector = ModelSelector()
+        self.ensemble_aggregator = EnsembleAggregator(min_models_required=min_models_required)
+        self.percentile_calculator = PercentileCalculator()
+
+        # Registry of implemented models
+        self.model_registry = {
+            ValuationModel.DCF: DCFModel(),
+            ValuationModel.DDM: DDMModel(),
+            ValuationModel.PE_MULTIPLE: PEMultipleModel(),
+            # TODO: Add remaining models as they're implemented
+        }
+
+    async def __aenter__(self):
+        """Async context manager entry"""
+        await self.data_fetcher.__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit"""
+        await self.data_fetcher.__aexit__(exc_type, exc_val, exc_tb)
+
+    async def value_stock(self,
+                         ticker: str,
+                         assumptions: Optional[ValuationAssumptions] = None) -> AggregatedValuation:
+        """
+        Perform comprehensive valuation of a stock.
+
+        Workflow:
+        1. Fetch comprehensive data
+        2. Select applicable models
+        3. Run models in parallel
+        4. Aggregate results
+        5. Calculate percentiles
+        6. Return AggregatedValuation
+
+        Args:
+            ticker: Stock ticker symbol
+            assumptions: Custom valuation assumptions (optional)
+
+        Returns:
+            AggregatedValuation with all results
+        """
+        # Use default assumptions if not provided
+        if assumptions is None:
+            assumptions = ValuationAssumptions()
+
+        # Step 1: Fetch comprehensive data
+        print(f"[ValuationEngine] Fetching data for {ticker}...")
+        stock_entity = await self.data_fetcher.fetch_comprehensive_data(ticker)
+
+        if not stock_entity:
+            return AggregatedValuation(
+                ticker=ticker,
+                current_price=0.0,
+                model_results={},
+                weighted_fair_value=None
+            )
+
+        print(f"[ValuationEngine] Data completeness: {stock_entity.data_completeness_score:.2f}")
+
+        # Step 2: Select applicable models
+        print(f"[ValuationEngine] Selecting applicable models...")
+        selected_models = self.model_selector.select_applicable_models(stock_entity)
+        print(f"[ValuationEngine] Selected {len(selected_models)} models:")
+        for model, confidence in selected_models.items():
+            print(f"  - {model.value}: confidence {confidence:.2f}")
+
+        # Step 3: Run applicable models in parallel
+        print(f"[ValuationEngine] Running models in parallel...")
+        model_results = await self._run_models_parallel(
+            stock_entity,
+            selected_models,
+            assumptions
+        )
+
+        # Count valid results
+        valid_count = sum(1 for result in model_results.values() if result.is_valid)
+        print(f"[ValuationEngine] {valid_count}/{len(model_results)} models succeeded")
+
+        # Step 4: Aggregate results
+        print(f"[ValuationEngine] Aggregating results...")
+        aggregation = self.ensemble_aggregator.aggregate(model_results)
+
+        # Step 5: Calculate percentiles
+        percentiles = self.percentile_calculator.calculate_percentiles(model_results)
+
+        # Step 6: Build AggregatedValuation object
+        aggregated_valuation = AggregatedValuation(
+            ticker=ticker,
+            current_price=stock_entity.price,
+            model_results=model_results,
+            weighted_fair_value=aggregation['weighted_fair_value'],
+            percentile_10=percentiles['percentile_10'],
+            percentile_50=percentiles['percentile_50'],
+            percentile_90=percentiles['percentile_90'],
+            valid_models_count=aggregation['valid_models_count'],
+            total_confidence=aggregation['total_confidence']
+        )
+
+        # Calculate upside/downside
+        aggregated_valuation.calculate_upside_downside()
+
+        print(f"[ValuationEngine] Valuation complete!")
+        print(f"  Current Price: ${stock_entity.price:.2f}")
+        if aggregated_valuation.weighted_fair_value:
+            print(f"  Fair Value: ${aggregated_valuation.weighted_fair_value:.2f}")
+            print(f"  Upside/Downside: {aggregated_valuation.upside_downside:.1f}%")
+        print(f"  Range (P10-P90): ${percentiles['percentile_10']:.2f} - ${percentiles['percentile_90']:.2f}")
+
+        return aggregated_valuation
+
+    async def _run_models_parallel(self,
+                                   stock_entity: StockEntity,
+                                   selected_models: Dict[ValuationModel, float],
+                                   assumptions: ValuationAssumptions) -> Dict[ValuationModel, ValuationResult]:
+        """
+        Run multiple valuation models in parallel.
+
+        Args:
+            stock_entity: StockEntity with company data
+            selected_models: Dictionary of selected models with confidence scores
+            assumptions: ValuationAssumptions
+
+        Returns:
+            Dictionary mapping ValuationModel to ValuationResult
+        """
+        tasks = []
+        model_types = []
+
+        # Create tasks for each model
+        for model_type, base_confidence in selected_models.items():
+            # Check if model is implemented
+            if model_type not in self.model_registry:
+                print(f"[WARNING] Model {model_type.value} not yet implemented, skipping...")
+                continue
+
+            model_instance = self.model_registry[model_type]
+
+            # Create task
+            task = model_instance.calculate_safe(stock_entity, assumptions)
+            tasks.append(task)
+            model_types.append(model_type)
+
+        # Run all models in parallel
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        # Build results dictionary
+        model_results = {}
+        for model_type, result in zip(model_types, results):
+            if isinstance(result, Exception):
+                # Handle exceptions
+                model_results[model_type] = ValuationResult(
+                    model=model_type,
+                    error=f"Model execution failed: {str(result)}",
+                    confidence=0.0
+                )
+            else:
+                model_results[model_type] = result
+
+        return model_results
+
+    async def get_quick_valuation(self, ticker: str) -> Optional[float]:
+        """
+        Get quick fair value estimate (weighted average only).
+
+        Faster alternative to full valuation when you just need a number.
+
+        Args:
+            ticker: Stock ticker symbol
+
+        Returns:
+            Weighted fair value or None
+        """
+        result = await self.value_stock(ticker)
+        return result.weighted_fair_value
+
+    # TODO: Implement risk analysis methods
+    # async def calculate_wacc(self, stock_entity: StockEntity) -> float:
+    #     """Calculate WACC for discount rate"""
+    #     pass
+
+    # async def run_monte_carlo(self, stock_entity: StockEntity, iterations: int = 10000):
+    #     """Run Monte Carlo simulation"""
+    #     pass
+
+    # TODO: Implement reporting methods
+    # async def generate_report(self, ticker: str, output_format: str = "markdown") -> str:
+    #     """Generate comprehensive valuation report"""
+    #     pass

src/core/valuation_engine/selector/__init__.py

@@ -0,0 +1,5 @@
+"""Model selection logic"""
+
+from src.core.valuation_engine.selector.model_selector import ModelSelector
+
+__all__ = ["ModelSelector"]

src/core/valuation_engine/selector/model_selector.py

@@ -0,0 +1,256 @@
+"""Rule-based model selector with confidence scoring"""
+
+from typing import Dict, List
+from src.core.valuation_engine.core.models import ValuationModel
+from src.core.valuation_engine.core.stock_entity import StockEntity
+from src.core.valuation_engine.selector.selector_config import (
+    MODEL_SELECTION_RULES,
+    SECTOR_MODEL_PREFERENCES
+)
+
+
+class ModelSelector:
+    """
+    Rule-based model selector that determines which valuation models
+    are applicable to a given stock and assigns confidence scores.
+    """
+
+    def __init__(self):
+        pass
+
+    def select_applicable_models(self, stock: StockEntity) -> Dict[ValuationModel, float]:
+        """
+        Select applicable models and assign base confidence scores.
+
+        Args:
+            stock: StockEntity with company data
+
+        Returns:
+            Dictionary mapping ValuationModel to confidence score (0-1)
+        """
+        applicable_models = {}
+
+        # Iterate through all models
+        for model, rules in MODEL_SELECTION_RULES.items():
+            # Check if model meets requirements
+            if self._check_requirements(stock, rules["requires"]):
+                # Calculate confidence score
+                confidence = self._calculate_model_confidence(stock, model, rules)
+
+                # Only include if confidence meets minimum
+                if confidence >= rules["min_confidence"]:
+                    applicable_models[model] = confidence
+
+        return applicable_models
+
+    def _check_requirements(self, stock: StockEntity, requirements: List[str]) -> bool:
+        """
+        Check if stock meets all requirements for a model.
+
+        Args:
+            stock: StockEntity
+            requirements: List of requirement strings (e.g., "free_cash_flow > 0")
+
+        Returns:
+            True if all requirements met
+        """
+        for req in requirements:
+            if not self._evaluate_requirement(stock, req):
+                return False
+        return True
+
+    def _evaluate_requirement(self, stock: StockEntity, requirement: str) -> bool:
+        """
+        Evaluate a single requirement string.
+
+        Args:
+            stock: StockEntity
+            requirement: Requirement string (e.g., "eps > 0")
+
+        Returns:
+            True if requirement is met
+        """
+        # Parse requirement (format: "field operator value")
+        parts = requirement.split()
+        if len(parts) != 3:
+            return False
+
+        field_name, operator, threshold_str = parts
+
+        # Get field value from stock
+        value = getattr(stock, field_name, None)
+
+        # If field doesn't exist or is None, requirement not met
+        if value is None:
+            return False
+
+        # Convert threshold to appropriate type
+        try:
+            threshold = float(threshold_str)
+        except ValueError:
+            threshold = threshold_str
+
+        # Evaluate operator
+        if operator == ">":
+            return value > threshold
+        elif operator == ">=":
+            return value >= threshold
+        elif operator == "<":
+            return value < threshold
+        elif operator == "<=":
+            return value <= threshold
+        elif operator == "==":
+            return value == threshold
+        elif operator == "!=":
+            return value != threshold
+        else:
+            return False
+
+    def _calculate_model_confidence(self,
+                                    stock: StockEntity,
+                                    model: ValuationModel,
+                                    rules: dict) -> float:
+        """
+        Calculate confidence score for a model.
+
+        Factors:
+        - Base confidence (from rules)
+        - Sector match
+        - Market cap match
+        - Data completeness
+        - Sector-specific preferences
+
+        Args:
+            stock: StockEntity
+            model: ValuationModel
+            rules: Model rules dictionary
+
+        Returns:
+            Confidence score (0-1)
+        """
+        base_confidence = rules["base_confidence"]
+
+        # Factor 1: Sector preference
+        sector_factor = 1.0
+        if stock.sector:
+            preferred_sectors = rules.get("preferred_sectors", [])
+
+            if preferred_sectors:  # If specific sectors are preferred
+                if stock.sector in preferred_sectors:
+                    sector_factor = 1.10  # 10% bonus for preferred sector
+                else:
+                    sector_factor = 0.95  # 5% penalty if not preferred
+
+            # Check sector-specific model rankings
+            sector_prefs = SECTOR_MODEL_PREFERENCES.get(stock.sector, [])
+            if model in sector_prefs:
+                # Higher ranking = higher confidence
+                rank = sector_prefs.index(model) + 1
+                # Top 3 models get bonus
+                if rank <= 3:
+                    sector_factor *= 1.05
+
+        # Factor 2: Market cap preference
+        market_cap_factor = 1.0
+        if stock.market_cap:
+            market_cap_category = stock.get_market_cap_category()
+            preferred_caps = rules.get("preferred_market_caps", [])
+
+            if preferred_caps and market_cap_category in preferred_caps:
+                market_cap_factor = 1.05  # 5% bonus
+
+        # Factor 3: Data completeness
+        completeness_score = stock.calculate_completeness_score()
+        data_factor = 0.8 + 0.2 * completeness_score  # Range: 0.8 to 1.0
+
+        # Factor 4: Model-specific adjustments
+        model_specific_factor = self._get_model_specific_factor(stock, model)
+
+        # Calculate final confidence
+        confidence = (base_confidence *
+                     sector_factor *
+                     market_cap_factor *
+                     data_factor *
+                     model_specific_factor)
+
+        # Ensure confidence is in [0, 1] range
+        return max(0.0, min(1.0, confidence))
+
+    def _get_model_specific_factor(self, stock: StockEntity, model: ValuationModel) -> float:
+        """
+        Get model-specific confidence adjustments based on stock characteristics.
+
+        Args:
+            stock: StockEntity
+            model: ValuationModel
+
+        Returns:
+            Adjustment factor (typically 0.8 to 1.2)
+        """
+        factor = 1.0
+
+        if model == ValuationModel.DCF:
+            # DCF works better with consistent FCF history
+            if stock.historical_fcf:
+                valid_fcf = [f for f in stock.historical_fcf if f is not None and f > 0]
+                if len(valid_fcf) >= 4:
+                    factor *= 1.10  # Good FCF history
+
+        elif model == ValuationModel.DDM:
+            # DDM works better for dividend aristocrats
+            if stock.historical_dividends:
+                valid_dividends = [d for d in stock.historical_dividends if d is not None and d > 0]
+                if len(valid_dividends) >= 3:
+                    # Check for consistent dividend growth
+                    no_cuts = all(valid_dividends[i] <= valid_dividends[i-1]
+                                for i in range(1, len(valid_dividends)))
+                    if no_cuts:
+                        factor *= 1.15  # Dividend aristocrat
+
+        elif model == ValuationModel.PE_MULTIPLE:
+            # P/E works better for stable earnings
+            if stock.financial_metrics and stock.financial_metrics.quality_of_earnings:
+                if stock.financial_metrics.quality_of_earnings > 1.0:
+                    factor *= 1.05  # High quality earnings
+
+        elif model == ValuationModel.ASSET_BASED:
+            # Asset-based works better for asset-heavy industries
+            if stock.is_financial_company:
+                factor *= 1.15  # Banks, REITs, etc.
+
+        elif model == ValuationModel.PEG:
+            # PEG works better for growth companies
+            if stock.revenue_growth and stock.revenue_growth > 0.10:
+                factor *= 1.10  # High growth
+
+        elif model == ValuationModel.RESIDUAL_INCOME:
+            # RIM works better for financial companies
+            if stock.is_financial_company:
+                factor *= 1.12
+
+        return factor
+
+    def get_recommended_models(self,
+                              stock: StockEntity,
+                              max_models: int = 5) -> List[ValuationModel]:
+        """
+        Get list of recommended models sorted by confidence.
+
+        Args:
+            stock: StockEntity
+            max_models: Maximum number of models to return
+
+        Returns:
+            List of ValuationModels sorted by confidence (descending)
+        """
+        applicable_models = self.select_applicable_models(stock)
+
+        # Sort by confidence (descending)
+        sorted_models = sorted(
+            applicable_models.items(),
+            key=lambda x: x[1],
+            reverse=True
+        )
+
+        # Return top N models
+        return [model for model, _ in sorted_models[:max_models]]

src/core/valuation_engine/selector/selector_config.py

@@ -0,0 +1,171 @@
+"""Configuration for model selection rules"""
+
+from src.core.valuation_engine.core.models import ValuationModel
+
+
+# Model selection rules
+MODEL_SELECTION_RULES = {
+    ValuationModel.DCF: {
+        "requires": ["free_cash_flow > 0", "shares_outstanding > 0"],
+        "preferred_sectors": ["Technology", "Healthcare", "Consumer Cyclical", "Industrials"],
+        "preferred_market_caps": ["large", "mid"],
+        "base_confidence": 0.80,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.DDM: {
+        "requires": ["dividend_per_share > 0"],
+        "preferred_sectors": ["Utilities", "Financial Services", "Consumer Defensive", "Real Estate"],
+        "preferred_market_caps": ["large", "mid"],
+        "base_confidence": 0.75,
+        "min_confidence": 0.40,
+    },
+    ValuationModel.PE_MULTIPLE: {
+        "requires": ["eps > 0", "price > 0"],
+        "preferred_sectors": [],  # Applicable to all sectors
+        "preferred_market_caps": ["large", "mid", "small"],
+        "base_confidence": 0.70,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.PEG: {
+        "requires": ["eps > 0", "earnings_growth > 0"],
+        "preferred_sectors": ["Technology", "Healthcare", "Consumer Cyclical"],
+        "preferred_market_caps": ["mid", "small"],
+        "base_confidence": 0.70,
+        "min_confidence": 0.35,
+    },
+    ValuationModel.EV_EBITDA: {
+        "requires": ["ebitda > 0"],
+        "preferred_sectors": ["Industrials", "Basic Materials", "Energy", "Technology"],
+        "preferred_market_caps": ["large", "mid"],
+        "base_confidence": 0.72,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.EV_SALES: {
+        "requires": ["revenue > 0"],
+        "preferred_sectors": ["Technology", "Healthcare"],  # Often for growth companies
+        "preferred_market_caps": ["mid", "small"],
+        "base_confidence": 0.60,
+        "min_confidence": 0.25,
+    },
+    ValuationModel.RESIDUAL_INCOME: {
+        "requires": ["book_value > 0", "roe > 0"],
+        "preferred_sectors": ["Financial Services", "Financial", "Real Estate"],
+        "preferred_market_caps": ["large", "mid"],
+        "base_confidence": 0.75,
+        "min_confidence": 0.35,
+    },
+    ValuationModel.ASSET_BASED: {
+        "requires": ["total_assets > 0", "total_liabilities >= 0"],
+        "preferred_sectors": ["Financial Services", "Real Estate", "Industrials"],
+        "preferred_market_caps": ["large", "mid", "small"],
+        "base_confidence": 0.68,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.GRAHAM: {
+        "requires": ["eps > 0", "book_value_per_share > 0"],
+        "preferred_sectors": ["Consumer Defensive", "Utilities", "Industrials"],
+        "preferred_market_caps": ["large", "mid"],
+        "base_confidence": 0.65,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.REVERSE_DCF: {
+        "requires": ["price > 0", "free_cash_flow > 0"],
+        "preferred_sectors": [],  # Applicable to all
+        "preferred_market_caps": ["large", "mid", "small"],
+        "base_confidence": 0.68,
+        "min_confidence": 0.25,
+    },
+    ValuationModel.SCENARIO_BASED: {
+        "requires": ["free_cash_flow > 0"],
+        "preferred_sectors": [],  # Applicable to all
+        "preferred_market_caps": ["large", "mid", "small"],
+        "base_confidence": 0.72,
+        "min_confidence": 0.30,
+    },
+    ValuationModel.MARKET_IMPLIED: {
+        "requires": ["price > 0"],
+        "preferred_sectors": [],  # Applicable to all
+        "preferred_market_caps": ["large", "mid", "small"],
+        "base_confidence": 0.60,
+        "min_confidence": 0.25,
+    },
+}
+
+
+# Sector-specific model preferences (models ordered by preference)
+SECTOR_MODEL_PREFERENCES = {
+    "Technology": [
+        ValuationModel.DCF,
+        ValuationModel.PEG,
+        ValuationModel.EV_SALES,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.SCENARIO_BASED,
+    ],
+    "Healthcare": [
+        ValuationModel.DCF,
+        ValuationModel.PEG,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.EV_SALES,
+    ],
+    "Financial Services": [
+        ValuationModel.RESIDUAL_INCOME,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.ASSET_BASED,
+        ValuationModel.DDM,
+    ],
+    "Financial": [
+        ValuationModel.RESIDUAL_INCOME,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.ASSET_BASED,
+        ValuationModel.DDM,
+    ],
+    "Consumer Defensive": [
+        ValuationModel.DDM,
+        ValuationModel.GRAHAM,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.DCF,
+    ],
+    "Utilities": [
+        ValuationModel.DDM,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.DCF,
+        ValuationModel.GRAHAM,
+    ],
+    "Real Estate": [
+        ValuationModel.DDM,
+        ValuationModel.ASSET_BASED,
+        ValuationModel.RESIDUAL_INCOME,
+        ValuationModel.PE_MULTIPLE,
+    ],
+    "Industrials": [
+        ValuationModel.DCF,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.EV_EBITDA,
+        ValuationModel.ASSET_BASED,
+    ],
+    "Energy": [
+        ValuationModel.DCF,
+        ValuationModel.EV_EBITDA,
+        ValuationModel.PE_MULTIPLE,
+    ],
+    "Basic Materials": [
+        ValuationModel.DCF,
+        ValuationModel.EV_EBITDA,
+        ValuationModel.PE_MULTIPLE,
+    ],
+    "Consumer Cyclical": [
+        ValuationModel.DCF,
+        ValuationModel.PEG,
+        ValuationModel.PE_MULTIPLE,
+    ],
+    "Communication Services": [
+        ValuationModel.DCF,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.EV_SALES,
+    ],
+    "Communication": [
+        ValuationModel.DCF,
+        ValuationModel.PE_MULTIPLE,
+        ValuationModel.EV_SALES,
+    ],
+}