Add risk-adjusted volatility allocation analysis

App/analysis/portfolio_optimizer.py

@@ -1,12 +1,13 @@
 import asyncio
-import math
 from dataclasses import dataclass
 from datetime import date, timedelta
 from decimal import Decimal, InvalidOperation
+import re
 from typing import Any
 
 import numpy as np
 
+from App.analysis.volatility import calculate_annualized_volatility
 from App.routers.bonds.models import Bond
 from App.routers.funds.models import FundPerformance, MutualFund
 from App.routers.stocks.metrics import calculate_metrics
@@ -20,6 +21,32 @@ GROWTH_CLASS_LIMITS = {
     "BOND": (0.05, 0.35),
 }
 MAX_ASSET_WEIGHT = 0.35
+RISK_PROFILES = {
+    "conservative": {
+        "label": "Conservative",
+        "class_limits": {"STOCK": (0.00, 0.35), "FUND": (0.25, 0.70), "BOND": (0.15, 0.60)},
+        "max_asset_weight": 0.25,
+        "return_weight": 0.55,
+        "volatility_penalty": 1.35,
+        "income_weight": 0.45,
+    },
+    "balanced": {
+        "label": "Balanced",
+        "class_limits": GROWTH_CLASS_LIMITS,
+        "max_asset_weight": MAX_ASSET_WEIGHT,
+        "return_weight": 0.85,
+        "volatility_penalty": 0.85,
+        "income_weight": 0.25,
+    },
+    "growth": {
+        "label": "Growth",
+        "class_limits": {"STOCK": (0.30, 0.75), "FUND": (0.05, 0.45), "BOND": (0.00, 0.25)},
+        "max_asset_weight": 0.40,
+        "return_weight": 1.15,
+        "volatility_penalty": 0.45,
+        "income_weight": 0.10,
+    },
+}
 DEFAULT_RISK_FREE_RATE = 0.12
 DEFAULT_SIMULATIONS = 6000
 
@@ -36,9 +63,53 @@ class OptimizerAsset:
     current_value: float
     expected_return: float
     volatility: float
+    income_yield: float
     fee_rate: float
 
 
+def _blend_pair(start: tuple[float, float], end: tuple[float, float], t: float) -> tuple[float, float]:
+    return (
+        start[0] + (end[0] - start[0]) * t,
+        start[1] + (end[1] - start[1]) * t,
+    )
+
+
+def _risk_settings(risk_profile: str | None, risk_score: int | None = None) -> dict[str, Any]:
+    if risk_score is not None:
+        score = max(0, min(100, int(risk_score)))
+        t = score / 100
+        low = RISK_PROFILES["conservative"]
+        high = RISK_PROFILES["growth"]
+        if score < 34:
+            label = "Conservative"
+        elif score > 66:
+            label = "Aggressive"
+        else:
+            label = "Balanced"
+        return {
+            "key": "conservative" if score < 34 else "growth" if score > 66 else "balanced",
+            "label": label,
+            "risk_score": score,
+            "class_limits": {
+                asset_type: _blend_pair(low["class_limits"][asset_type], high["class_limits"][asset_type], t)
+                for asset_type in GROWTH_CLASS_LIMITS
+            },
+            "max_asset_weight": low["max_asset_weight"] + (high["max_asset_weight"] - low["max_asset_weight"]) * t,
+            "return_weight": low["return_weight"] + (high["return_weight"] - low["return_weight"]) * t,
+            "volatility_penalty": low["volatility_penalty"] + (high["volatility_penalty"] - low["volatility_penalty"]) * t,
+            "income_weight": low["income_weight"] + (high["income_weight"] - low["income_weight"]) * t,
+        }
+
+    key = str(risk_profile or "balanced").strip().lower()
+    if key in {"low", "safe", "safer", "defensive"}:
+        key = "conservative"
+    elif key in {"high", "aggressive"}:
+        key = "growth"
+    settings = RISK_PROFILES.get(key, RISK_PROFILES["balanced"])
+    fallback_score = 15 if key == "conservative" else 85 if key == "growth" else 50
+    return {"key": key if key in RISK_PROFILES else "balanced", "risk_score": fallback_score, **settings}
+
+
 def _safe_float(value: Any, default: float = 0.0) -> float:
     if value in (None, ""):
         return default
@@ -59,16 +130,6 @@ def _annualized_return(values: list[tuple[date, float]]) -> float | None:
     return (1 + total_return) ** (365 / days) - 1
 
 
-def _annualized_volatility(values: list[tuple[date, float]]) -> float | None:
-    ordered = [v for _, v in sorted(values, key=lambda item: item[0]) if v > 0]
-    if len(ordered) < 3:
-        return None
-    returns = np.diff(np.log(np.array(ordered, dtype=float)))
-    if len(returns) == 0:
-        return None
-    return float(np.std(returns) * math.sqrt(252))
-
-
 def _stock_fee_rate(consideration: float) -> float:
     payload = load_dse_transaction_fee_seed_data()
     for band in payload.get("bands", []):
@@ -88,6 +149,33 @@ def _parse_percent(raw_value: Any) -> float:
     return _safe_float(digits) / 100 if digits else 0.0
 
 
+def _parse_distribution_rate(raw_value: Any) -> float:
+    text = str(raw_value or "")
+    matches = re.findall(r"(\d+(?:\.\d+)?)\s*%", text)
+    if not matches:
+        return 0.0
+    return max(_safe_float(match) / 100 for match in matches)
+
+
+async def _fund_income_yield(fund: MutualFund) -> float:
+    if not fund.pays_income:
+        return 0.0
+    try:
+        info = await fund.info_record
+    except Exception:
+        info = None
+    data = info.raw_data if info and isinstance(info.raw_data, dict) else {}
+    distribution = data.get("distribution") if isinstance(data.get("distribution"), dict) else {}
+    other_facts = data.get("other_facts") if isinstance(data.get("other_facts"), dict) else {}
+    candidates = [
+        fund.income_amount,
+        distribution.get("max_distribution_rate"),
+        distribution.get("policy"),
+        other_facts.get("max_distribution_rate"),
+    ]
+    return min(max((_parse_distribution_rate(value) for value in candidates), default=0.0), 0.18)
+
+
 async def _collect_positions(portfolio_id: int) -> list[dict[str, Any]]:
     from App.routers.portfolio.service import PortfolioService
 
@@ -110,7 +198,7 @@ async def _build_stock_asset(position: dict[str, Any]) -> OptimizerAsset | None:
 
     prices = await StockPriceData.filter(
         stock_id=stock.id,
-        date__gte=date.today() - timedelta(days=365 * 3),
+        date__gte=date.today() - timedelta(days=365 * 5 + 7),
     ).order_by("date").values("date", "closing_price")
     price_values = [(row["date"], _safe_float(row["closing_price"])) for row in prices]
     latest = await StockPriceData.filter(stock=stock).order_by("-date").first()
@@ -121,7 +209,10 @@ async def _build_stock_asset(position: dict[str, Any]) -> OptimizerAsset | None:
     capital_return = _annualized_return(price_values)
     dividend_yield = _safe_float(metrics.get("dividend_yield")) / 100
     expected_return = (capital_return if capital_return is not None else 0.10) + dividend_yield
-    volatility = _annualized_volatility(price_values) or 0.28
+    volatility_result = calculate_annualized_volatility(price_values)
+    volatility = (
+        volatility_result.annualized_volatility if volatility_result else 0.28
+    )
     current_price = _safe_float(latest.closing_price)
     quantity = _safe_float(position["quantity"])
     current_value = quantity * current_price
@@ -137,6 +228,7 @@ async def _build_stock_asset(position: dict[str, Any]) -> OptimizerAsset | None:
         current_value=current_value,
         expected_return=max(min(expected_return, 0.60), -0.30),
         volatility=max(volatility, 0.08),
+        income_yield=max(dividend_yield, 0.0),
         fee_rate=_stock_fee_rate(current_value),
     )
 
@@ -148,7 +240,7 @@ async def _build_fund_asset(position: dict[str, Any]) -> OptimizerAsset | None:
 
     rows = await FundPerformance.filter(
         fund_id=fund.id,
-        record_date__gte=date.today() - timedelta(days=365 * 3),
+        record_date__gte=date.today() - timedelta(days=365 * 5 + 7),
     ).order_by("record_date").values("record_date", "nav_per_unit")
     nav_values = [
         (row["record_date"], _safe_float(row["nav_per_unit"]))
@@ -159,8 +251,12 @@ async def _build_fund_asset(position: dict[str, Any]) -> OptimizerAsset | None:
     if latest is None or latest.nav_per_unit is None:
         return None
 
-    expected_return = _annualized_return(nav_values) or 0.13
-    volatility = _annualized_volatility(nav_values) or 0.08
+    income_yield = await _fund_income_yield(fund)
+    expected_return = (_annualized_return(nav_values) or 0.13) + income_yield
+    volatility_result = calculate_annualized_volatility(nav_values)
+    volatility = (
+        volatility_result.annualized_volatility if volatility_result else 0.08
+    )
     current_price = _safe_float(latest.nav_per_unit)
     quantity = _safe_float(position["quantity"])
     current_value = quantity * current_price
@@ -176,6 +272,7 @@ async def _build_fund_asset(position: dict[str, Any]) -> OptimizerAsset | None:
         current_value=current_value,
         expected_return=max(min(expected_return, 0.35), 0.03),
         volatility=max(volatility, 0.03),
+        income_yield=income_yield,
         fee_rate=_safe_float(fund.entry_load) / 100,
     )
 
@@ -202,6 +299,7 @@ async def _build_bond_asset(position: dict[str, Any]) -> OptimizerAsset | None:
         current_value=current_value,
         expected_return=max(coupon_rate, 0.10),
         volatility=0.04,
+        income_yield=coupon_rate,
         fee_rate=0.0,
     )
 
@@ -223,6 +321,81 @@ async def _build_assets(portfolio_id: int) -> list[OptimizerAsset]:
     return assets
 
 
+async def _stock_candidate(stock: Stock) -> dict[str, Any] | None:
+    asset = await _build_stock_asset({"asset_id": stock.id, "quantity": Decimal("1")})
+    if asset is None:
+        return None
+    return {
+        "asset_id": asset.asset_id,
+        "asset_type": asset.asset_type,
+        "symbol": asset.symbol,
+        "name": asset.name,
+        "expected_return": round(asset.expected_return, 4),
+        "volatility": round(asset.volatility, 4),
+        "income_yield": round(asset.income_yield, 4),
+        "reason": "Lower-volatility stock candidate with available price and dividend history.",
+    }
+
+
+async def _fund_candidate(fund: MutualFund) -> dict[str, Any] | None:
+    asset = await _build_fund_asset({"asset_id": fund.id, "quantity": Decimal("1")})
+    if asset is None:
+        return None
+    return {
+        "asset_id": asset.asset_id,
+        "asset_type": asset.asset_type,
+        "symbol": asset.symbol,
+        "name": asset.name,
+        "expected_return": round(asset.expected_return, 4),
+        "volatility": round(asset.volatility, 4),
+        "income_yield": round(asset.income_yield, 4),
+        "pays_income": asset.income_yield > 0,
+        "reason": (
+            "Income-paying fund candidate with lower measured volatility."
+            if asset.income_yield > 0
+            else "Fund candidate with lower measured volatility."
+        ),
+    }
+
+
+async def _suggest_alternatives(current_assets: list[OptimizerAsset], risk_profile: str | None, risk_score: int | None) -> list[dict[str, Any]]:
+    current_keys = {asset.key for asset in current_assets}
+    settings = _risk_settings(risk_profile, risk_score)
+    candidates: list[dict[str, Any]] = []
+
+    funds = await MutualFund.filter(status="Active").all()
+    for fund in funds:
+        if f"FUND:{fund.id}" in current_keys:
+            continue
+        candidate = await _fund_candidate(fund)
+        if candidate:
+            candidates.append(candidate)
+
+    stocks = await Stock.all()
+    for stock in stocks:
+        if f"STOCK:{stock.id}" in current_keys:
+            continue
+        candidate = await _stock_candidate(stock)
+        if candidate:
+            candidates.append(candidate)
+
+    if not candidates:
+        return []
+
+    current_volatility = max((asset.volatility for asset in current_assets), default=0.0)
+    max_volatility = current_volatility * (0.85 if settings["key"] == "conservative" else 1.10)
+    safer = [item for item in candidates if item["volatility"] <= max_volatility] or candidates
+
+    def score(item: dict[str, Any]) -> float:
+        return (
+            item["expected_return"] * settings["return_weight"]
+            + item.get("income_yield", 0.0) * settings["income_weight"]
+            - item["volatility"] * settings["volatility_penalty"]
+        )
+
+    return sorted(safer, key=score, reverse=True)[:5]
+
+
 def _portfolio_stats(weights: np.ndarray, returns: np.ndarray, vols: np.ndarray) -> dict:
     portfolio_return = float(np.dot(weights, returns))
     portfolio_vol = float(np.sqrt(np.dot(weights**2, vols**2)))
@@ -245,8 +418,8 @@ def _class_totals(weights: np.ndarray, assets: list[OptimizerAsset]) -> dict[str
     return totals
 
 
-def _respects_limits(weights: np.ndarray, assets: list[OptimizerAsset]) -> bool:
-    effective_max_asset_weight = max(MAX_ASSET_WEIGHT, 1 / len(assets))
+def _respects_limits(weights: np.ndarray, assets: list[OptimizerAsset], settings: dict[str, Any]) -> bool:
+    effective_max_asset_weight = max(float(settings["max_asset_weight"]), 1 / len(assets))
     if np.any(weights < 0) or np.any(weights > effective_max_asset_weight):
         return False
     available_classes = {asset.asset_type for asset in assets}
@@ -254,9 +427,10 @@ def _respects_limits(weights: np.ndarray, assets: list[OptimizerAsset]) -> bool:
         return True
 
     class_totals = _class_totals(weights, assets)
+    class_limits = settings["class_limits"]
     for asset_type in available_classes:
         total = class_totals.get(asset_type, 0.0)
-        min_weight, max_weight = GROWTH_CLASS_LIMITS[asset_type]
+        min_weight, max_weight = class_limits[asset_type]
         if total < min_weight or total > max_weight:
             return False
     return True
@@ -268,11 +442,52 @@ def _estimate_rebalance_fees(
     assets: list[OptimizerAsset],
     total_value: float,
 ) -> float:
-    fees = 0.0
+    return sum(item["estimated_fee"] for item in _fee_breakdown(current_weights, suggested_weights, assets, total_value))
+
+
+def _fee_breakdown(
+    current_weights: np.ndarray,
+    suggested_weights: np.ndarray,
+    assets: list[OptimizerAsset],
+    total_value: float,
+) -> list[dict[str, Any]]:
+    items = []
     for current, suggested, asset in zip(current_weights, suggested_weights, assets):
         increase = max(0.0, float(suggested - current)) * total_value
-        fees += increase * asset.fee_rate
-    return fees
+        if increase <= 0 or asset.fee_rate <= 0:
+            continue
+        fee = increase * asset.fee_rate
+        items.append(
+            {
+                "asset_id": asset.asset_id,
+                "asset_type": asset.asset_type,
+                "symbol": asset.symbol,
+                "name": asset.name,
+                "buy_amount": round(increase, 2),
+                "fee_rate": round(asset.fee_rate, 6),
+                "estimated_fee": round(fee, 2),
+                "fee_source": (
+                    "DSE transaction fee band"
+                    if asset.asset_type == "STOCK"
+                    else "Fund entry load"
+                    if asset.asset_type == "FUND"
+                    else "No explicit buy fee"
+                ),
+            }
+        )
+    return sorted(items, key=lambda item: item["estimated_fee"], reverse=True)
+
+
+def _fee_summary(fee_items: list[dict[str, Any]]) -> dict[str, Any]:
+    by_type: dict[str, float] = {}
+    for item in fee_items:
+        by_type[item["asset_type"]] = by_type.get(item["asset_type"], 0.0) + item["estimated_fee"]
+    total = sum(by_type.values())
+    return {
+        "total": round(total, 2),
+        "by_asset_type": {key: round(value, 2) for key, value in by_type.items()},
+        "items": fee_items,
+    }
 
 
 def _advisor_comment(
@@ -280,6 +495,7 @@ def _advisor_comment(
     suggested_stats: dict[str, float],
     allocations: list[dict[str, Any]],
     estimated_fees: float,
+    risk_label: str,
 ) -> dict[str, Any]:
     increases = [item for item in allocations if item["difference"] > 0.03]
     reductions = [item for item in allocations if item["difference"] < -0.03]
@@ -289,7 +505,7 @@ def _advisor_comment(
     return_delta = suggested_stats["expected_return"] - current_stats["expected_return"]
 
     summary = (
-        "The suggested allocation keeps the portfolio growth-focused while trying "
+        f"The suggested allocation is tuned for a {risk_label.lower()} risk setting while trying "
         "to avoid one position carrying too much of the risk."
     )
     if sharpe_delta > 0.05:
@@ -325,17 +541,25 @@ def _advisor_comment(
         "key_actions": actions[:3],
         "caution": (
             "This is an allocation model, not a guarantee. It uses historical prices, NAVs, "
-            "bond coupons, and known transaction costs; financial statement quality and future news still need human review."
+            "dividends, fund payout flags, bond coupons, and known transaction costs; financial statement quality and future news still need human review."
         ),
     }
 
 
-def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> dict[str, Any]:
+def _optimize_sync(
+    assets_payload: list[dict[str, Any]],
+    simulations: int,
+    risk_profile: str | None,
+    risk_score: int | None,
+    alternatives: list[dict[str, Any]],
+) -> dict[str, Any]:
     assets = [OptimizerAsset(**payload) for payload in assets_payload]
+    settings = _risk_settings(risk_profile, risk_score)
     total_value = sum(asset.current_value for asset in assets)
     current_weights = np.array([asset.current_value / total_value for asset in assets])
     returns = np.array([asset.expected_return for asset in assets])
     vols = np.array([asset.volatility for asset in assets])
+    income_yields = np.array([asset.income_yield for asset in assets])
 
     best_weights = current_weights.copy()
     best_score = -float("inf")
@@ -344,12 +568,19 @@ def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> di
 
     for _ in range(attempts):
         weights = rng.dirichlet(np.ones(len(assets)))
-        if not _respects_limits(weights, assets):
+        if not _respects_limits(weights, assets, settings):
             continue
         stats = _portfolio_stats(weights, returns, vols)
         fees = _estimate_rebalance_fees(current_weights, weights, assets, total_value)
         fee_drag = fees / total_value if total_value else 0.0
-        score = stats["sharpe"] + stats["expected_return"] - fee_drag
+        income_score = float(np.dot(weights, income_yields))
+        score = (
+            stats["sharpe"]
+            + settings["return_weight"] * stats["expected_return"]
+            + settings["income_weight"] * income_score
+            - settings["volatility_penalty"] * stats["volatility"]
+            - fee_drag
+        )
         if score > best_score:
             best_score = score
             best_weights = weights
@@ -359,6 +590,8 @@ def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> di
     estimated_fees = _estimate_rebalance_fees(
         current_weights, best_weights, assets, total_value
     )
+    fee_items = _fee_breakdown(current_weights, best_weights, assets, total_value)
+    fee_summary = _fee_summary(fee_items)
 
     allocations = []
     for asset, current_weight, suggested_weight in zip(
@@ -384,6 +617,7 @@ def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> di
                 "rebalance_amount": round(difference * total_value, 2),
                 "expected_return": round(asset.expected_return, 4),
                 "volatility": round(asset.volatility, 4),
+                "income_yield": round(asset.income_yield, 4),
                 "fee_rate": round(asset.fee_rate, 4),
                 "reason": reason,
             }
@@ -397,27 +631,33 @@ def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> di
 
     return {
         "objective": "growth",
+        "risk_profile": settings["key"],
+        "risk_label": settings["label"],
+        "risk_score": settings["risk_score"],
         "total_value": round(total_value, 2),
         "risk_free_rate": DEFAULT_RISK_FREE_RATE,
         "constraints": {
-            "max_asset_weight": max(MAX_ASSET_WEIGHT, 1 / len(assets)),
-            "class_limits": GROWTH_CLASS_LIMITS,
+            "max_asset_weight": max(float(settings["max_asset_weight"]), 1 / len(assets)),
+            "class_limits": settings["class_limits"],
             "class_limits_apply_only_when_multiple_asset_classes_exist": True,
         },
         "current": {k: round(v, 4) for k, v in current_stats.items()},
         "suggested": {k: round(v, 4) for k, v in suggested_stats.items()},
         "estimated_rebalance_fees": round(estimated_fees, 2),
+        "fee_breakdown": fee_summary,
         "advisor_comment": _advisor_comment(
             current_stats,
             suggested_stats,
             sorted_allocations,
             estimated_fees,
+            settings["label"],
         ),
         "allocations": sorted_allocations,
+        "alternatives": alternatives,
         "notes": [
-            "Uses current portfolio assets only; it does not yet recommend brand-new securities.",
+            "Suggested allocations rebalance current holdings; alternatives identify lower-volatility assets to research before adding new positions.",
             "Stock buy fee estimates use the stored DSE transaction fee bands.",
-            "Fund buy fee estimates use each fund's entry_load where available.",
+            "Stock dividends, fund income flags, bond coupons, and fund entry loads are included where available.",
             "Optimization uses Monte Carlo simulation to stay lightweight on Hugging Face.",
         ],
     }
@@ -426,10 +666,14 @@ def _optimize_sync(assets_payload: list[dict[str, Any]], simulations: int) -> di
 async def analyze_portfolio_growth_allocation(
     portfolio_id: int,
     simulations: int = DEFAULT_SIMULATIONS,
+    risk_profile: str | None = None,
+    risk_score: int | None = None,
 ) -> dict[str, Any]:
     assets = await _build_assets(portfolio_id)
     if not assets:
         raise ValueError("At least one priced asset is required for allocation analysis")
+    alternatives = await _suggest_alternatives(assets, risk_profile, risk_score)
+    settings = _risk_settings(risk_profile, risk_score)
     if len(assets) == 1:
         asset = assets[0]
         current_stats = {
@@ -459,6 +703,9 @@ async def analyze_portfolio_growth_allocation(
         }
         return {
             "objective": "growth",
+            "risk_profile": settings["key"],
+            "risk_label": settings["label"],
+            "risk_score": settings["risk_score"],
             "total_value": round(asset.current_value, 2),
             "risk_free_rate": DEFAULT_RISK_FREE_RATE,
             "constraints": {
@@ -469,6 +716,7 @@ async def analyze_portfolio_growth_allocation(
             "current": {k: round(v, 4) for k, v in current_stats.items()},
             "suggested": {k: round(v, 4) for k, v in current_stats.items()},
             "estimated_rebalance_fees": 0.0,
+            "fee_breakdown": _fee_summary([]),
             "advisor_comment": {
                 "title": "Concentration warning",
                 "summary": (
@@ -487,6 +735,7 @@ async def analyze_portfolio_growth_allocation(
                 ),
             },
             "allocations": [allocation],
+            "alternatives": alternatives,
             "notes": [
                 "Single-asset portfolios receive a concentration analysis instead of a failed optimization.",
                 "Add at least two priced assets to enable portfolio rebalancing advice.",
@@ -496,6 +745,9 @@ async def analyze_portfolio_growth_allocation(
         _optimize_sync,
         [asset.__dict__ for asset in assets],
         simulations,
+        risk_profile,
+        risk_score,
+        alternatives,
     )
 
 
@@ -503,16 +755,20 @@ async def run_portfolio_analysis_task(
     task_id: int,
     portfolio_id: int,
     simulations: int = DEFAULT_SIMULATIONS,
+    risk_profile: str | None = None,
+    risk_score: int | None = None,
 ) -> None:
     await ImportTask.filter(id=task_id).update(
         status=TaskStatus.RUNNING,
         details={
             "portfolio_id": portfolio_id,
+            "risk_profile": _risk_settings(risk_profile, risk_score)["key"],
+            "risk_score": _risk_settings(risk_profile, risk_score)["risk_score"],
             "status_message": "Portfolio growth allocation analysis is running",
         },
     )
     try:
-        result = await analyze_portfolio_growth_allocation(portfolio_id, simulations)
+        result = await analyze_portfolio_growth_allocation(portfolio_id, simulations, risk_profile, risk_score)
     except Exception as exc:
         await ImportTask.filter(id=task_id).update(
             status=TaskStatus.FAILED,

App/analysis/volatility.py

@@ -0,0 +1,69 @@
+import math
+from dataclasses import asdict, dataclass
+from datetime import date, timedelta
+from statistics import pstdev
+from typing import Iterable
+
+
+@dataclass(frozen=True)
+class VolatilityResult:
+    annualized_volatility: float
+    observations: int
+    return_observations: int
+    start_date: str
+    end_date: str
+    years_covered: float
+    requested_years: int
+    annualization_periods: float
+    used_partial_history: bool
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def calculate_annualized_volatility(
+    values: Iterable[tuple[date, float]],
+    *,
+    requested_years: int = 5,
+) -> VolatilityResult | None:
+    """Calculate annualized volatility from the available history up to requested_years.
+
+    The annualization factor is inferred from the average calendar gap between
+    observations, so sparse fund NAV histories do not get treated like daily data.
+    """
+    ordered = sorted((d, float(v)) for d, v in values if d and v and float(v) > 0)
+    if len(ordered) < 3:
+        return None
+
+    latest_date = ordered[-1][0]
+    cutoff = latest_date - timedelta(days=round(365.25 * requested_years))
+    window = [(d, v) for d, v in ordered if d >= cutoff]
+    if len(window) < 3:
+        window = ordered
+
+    log_returns = [
+        math.log(current / previous)
+        for (_, previous), (_, current) in zip(window, window[1:])
+        if previous > 0 and current > 0
+    ]
+    if len(log_returns) < 2:
+        return None
+
+    start_date = window[0][0]
+    end_date = window[-1][0]
+    span_days = max((end_date - start_date).days, 1)
+    average_gap_days = span_days / max(len(window) - 1, 1)
+    annualization_periods = 365.25 / max(average_gap_days, 1 / 365.25)
+    annualized = pstdev(log_returns) * math.sqrt(annualization_periods)
+
+    return VolatilityResult(
+        annualized_volatility=annualized,
+        observations=len(window),
+        return_observations=len(log_returns),
+        start_date=start_date.isoformat(),
+        end_date=end_date.isoformat(),
+        years_covered=span_days / 365.25,
+        requested_years=requested_years,
+        annualization_periods=annualization_periods,
+        used_partial_history=(span_days / 365.25) < requested_years * 0.95,
+    )

App/routers/funds/routes.py

@@ -3,6 +3,7 @@ from datetime import date, timedelta
 from typing import List, Optional
 
 from App.schemas import ResponseModel, AppException
+from App.analysis.volatility import calculate_annualized_volatility
 from .models import FundManager, MutualFund, FundPerformance, FundInfo
 from .seed import load_fund_seed_data, sync_fund_info_from_json
 from App.routers.users.utils import get_current_user
@@ -84,6 +85,25 @@ def _get_fund_info(fund_name: str) -> dict | None:
     return None
 
 
+async def _calculate_fund_volatility(fund_id: int) -> dict | None:
+    rows = await FundPerformance.filter(
+        fund_id=fund_id,
+        record_date__gte=date.today() - timedelta(days=365 * 5 + 7),
+        nav_per_unit__isnull=False,
+    ).order_by("record_date").values("record_date", "nav_per_unit")
+    result = calculate_annualized_volatility(
+        [(row["record_date"], float(row["nav_per_unit"])) for row in rows]
+    )
+    if result is None:
+        return None
+    payload = result.to_dict()
+    payload["annualized_volatility"] = round(payload["annualized_volatility"] * 100, 2)
+    payload["volatility_decimal"] = round(result.annualized_volatility, 6)
+    payload["years_covered"] = round(payload["years_covered"], 2)
+    payload["annualization_periods"] = round(payload["annualization_periods"], 2)
+    return payload
+
+
 # ── LIST ALL FUNDS (with latest NAV) ────────────────────────────────────────
 
 @router.get("", response_model=ResponseModel)
@@ -101,6 +121,7 @@ async def list_funds():
     for f in funds:
         latest = await FundPerformance.filter(fund_id=f.id).order_by("-record_date").first()
         static_info = await _serialize_fund_info(f)
+        volatility = await _calculate_fund_volatility(f.id)
         fund_list.append({
             "id": f.id,
             "name": f.name,
@@ -118,6 +139,10 @@ async def list_funds():
             "min_additional": f.min_additional,
             "benchmark": f.benchmark,
             "risk_level": static_info.get("risk_level") if static_info else None,
+            "five_year_volatility": volatility["annualized_volatility"] if volatility else None,
+            "annualized_volatility": volatility["annualized_volatility"] if volatility else None,
+            "volatility_decimal": volatility["volatility_decimal"] if volatility else None,
+            "volatility_window": volatility,
             "income_options": static_info.get("income_options", []) if static_info else [],
             "info": static_info,
         })
@@ -185,6 +210,7 @@ async def get_funds_performance():
         week_nav  = float(week_rec.nav_per_unit)  if week_rec  and week_rec.nav_per_unit  is not None else None
         month_nav = float(month_rec.nav_per_unit) if month_rec and month_rec.nav_per_unit is not None else None
         ytd_nav   = float(ytd_rec.nav_per_unit)   if ytd_rec   and ytd_rec.nav_per_unit   is not None else None
+        volatility = await _calculate_fund_volatility(f.id)
 
         result.append({
             "id":             f.id,
@@ -203,6 +229,10 @@ async def get_funds_performance():
             "weekly_return":  pct(week_nav,  curr),
             "monthly_return": pct(month_nav, curr),
             "ytd_return":     pct(ytd_nav,   curr),
+            "five_year_volatility": volatility["annualized_volatility"] if volatility else None,
+            "annualized_volatility": volatility["annualized_volatility"] if volatility else None,
+            "volatility_decimal": volatility["volatility_decimal"] if volatility else None,
+            "volatility_window": volatility,
         })
 
     result.sort(key=lambda x: x.get("monthly_return") or 0, reverse=True)

App/routers/portfolio/routes.py

@@ -36,6 +36,7 @@ from .models import (
 from .schemas import (
     PortfolioCreate,
     PortfolioAdvisorChatRequest,
+    GrowthAllocationAnalysisRequest,
     PortfolioUpdate,
     StockHoldingCreate,
     StockHoldingUpdate,
@@ -458,10 +459,16 @@ async def get_portfolio_summary(
 async def start_growth_allocation_analysis(
     portfolio_id: int,
     background_tasks: BackgroundTasks,
+    payload: GrowthAllocationAnalysisRequest | None = None,
     simulations: int = Query(6000, ge=1000, le=50000),
+    risk_profile: str = Query("balanced"),
+    risk_score: Optional[int] = Query(None, ge=0, le=100),
     current_user=Depends(get_current_user),
 ):
     await _verify_ownership(portfolio_id, current_user)
+    requested_simulations = payload.simulations if payload else simulations
+    requested_risk_profile = payload.risk_profile if payload else risk_profile
+    requested_risk_score = payload.risk_score if payload and payload.risk_score is not None else risk_score
 
     candidate_tasks = await ImportTask.filter(
         task_type="portfolio_growth_analysis",
@@ -473,6 +480,8 @@ async def start_growth_allocation_analysis(
             for task in candidate_tasks
             if isinstance(task.details, dict)
             and task.details.get("portfolio_id") == portfolio_id
+            and task.details.get("risk_profile", "balanced") == requested_risk_profile
+            and task.details.get("risk_score") == requested_risk_score
         ),
         None,
     )
@@ -492,7 +501,9 @@ async def start_growth_allocation_analysis(
         status=TaskStatus.PENDING,
         details={
             "portfolio_id": portfolio_id,
-            "simulations": simulations,
+            "simulations": requested_simulations,
+            "risk_profile": requested_risk_profile,
+            "risk_score": requested_risk_score,
             "status_message": "Portfolio growth allocation analysis has been queued",
         },
     )
@@ -500,7 +511,9 @@ async def start_growth_allocation_analysis(
         run_portfolio_analysis_task,
         task.id,
         portfolio_id,
-        simulations,
+        requested_simulations,
+        requested_risk_profile,
+        requested_risk_score,
     )
 
     return ResponseModel(

App/routers/portfolio/schemas.py

@@ -43,6 +43,12 @@ class PortfolioAdvisorChatRequest(BaseModel):
     hypothetical_portfolio: Optional[Dict[str, Any]] = None
 
 
+class GrowthAllocationAnalysisRequest(BaseModel):
+    simulations: int = Field(6000, ge=1000, le=50000)
+    risk_profile: str = Field("balanced", pattern="^(conservative|balanced|growth|low|safe|safer|defensive|high|aggressive)$")
+    risk_score: Optional[int] = Field(None, ge=0, le=100)
+
+
 # ──────────────── STOCK HOLDINGS ────────────────
 
 

App/routers/stocks/metrics.py

@@ -2,6 +2,8 @@ from datetime import date, timedelta
 from decimal import Decimal
 from statistics import mean, pstdev
 
+from App.analysis.volatility import calculate_annualized_volatility
+
 from .models import Dividend, StockPriceData, StockProfile
 
 
@@ -106,6 +108,14 @@ async def calculate_metrics(stock):
         if len(closing_prices) > 1
         else None
     )
+    volatility_result = calculate_annualized_volatility(
+        [
+            (row["date"], float(row["closing_price"]))
+            for row in all_rows
+            if row.get("closing_price") is not None
+        ]
+    )
+    volatility_payload = volatility_result.to_dict() if volatility_result else None
 
     # Fundamental metrics need processed statements; keep explicit nulls instead of fake ratios.
     eps = None
@@ -130,6 +140,22 @@ async def calculate_metrics(stock):
         else None,
         "return_percentage": _round(return_pct),
         "volatility": _round(volatility),
+        "five_year_volatility": (
+            _round(volatility_result.annualized_volatility * 100)
+            if volatility_result
+            else None
+        ),
+        "annualized_volatility": (
+            _round(volatility_result.annualized_volatility * 100)
+            if volatility_result
+            else None
+        ),
+        "volatility_decimal": (
+            round(volatility_result.annualized_volatility, 6)
+            if volatility_result
+            else None
+        ),
+        "volatility_window": volatility_payload,
         "average_market_cap": round(mean(market_caps)) if market_caps else None,
         "dividend_per_share": _round(dividend_per_share),
         "dividend_yield": _round(dividend_yield),