Protect high coupon bond income streams

App/analysis/portfolio_optimizer.py

@@ -51,6 +51,10 @@ RISK_PROFILES = {
 }
 DEFAULT_RISK_FREE_RATE = 0.12
 DEFAULT_SIMULATIONS = 6000
+HIGH_COUPON_ADVANTAGE_BPS = 150
+LEGACY_BOND_REPLACEMENT_MARGIN = 0.06
+LOW_MODERATE_VOLATILITY_MAX = 0.16
+MIN_REPLACEMENT_HISTORY_YEARS = 3.0
 
 
 @dataclass
@@ -135,6 +139,13 @@ def _annualized_return(values: list[tuple[date, float]]) -> float | None:
     return (1 + total_return) ** (365 / days) - 1
 
 
+def _history_years(values: list[tuple[date, float]]) -> float:
+    ordered = [item for item in sorted(values, key=lambda item: item[0]) if item[1] > 0]
+    if len(ordered) < 2:
+        return 0.0
+    return max((ordered[-1][0] - ordered[0][0]).days / 365.25, 0.0)
+
+
 def _stock_fee_band(consideration: float) -> dict[str, Any] | None:
     payload = load_dse_transaction_fee_seed_data()
     for band in payload.get("bands", []):
@@ -334,6 +345,7 @@ 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
     liquidity = metrics.get("liquidity") or await calculate_liquidity_metrics(stock)
+    liquidity["history_years"] = round(_history_years(price_values), 2)
     fundamentals = {
         "eps": metrics.get("eps"),
         "pe_ratio": metrics.get("pe_ratio"),
@@ -433,6 +445,7 @@ async def _build_fund_asset(position: dict[str, Any]) -> OptimizerAsset | None:
         liquidity={
             **prospectus_terms,
             "liquidity_score": round(max(liquidity_score, 20), 2),
+            "history_years": round(_history_years(nav_values), 2),
         },
     )
 
@@ -479,6 +492,11 @@ async def _build_bond_asset(position: dict[str, Any]) -> OptimizerAsset | None:
             "latest_yield_percent": latest_yield_percent,
             "coupon_rate_percent": _safe_float(bond.coupon_rate),
             "price_percent": price_per_100,
+            "primary_reinvestment_yield_percent": (
+                _safe_float((pricing_context.get("comparable_primary_market") or [{}])[0].get("implied_yield_percent"))
+                if pricing_context.get("comparable_primary_market")
+                else None
+            ),
             "same_bond_secondary_available": pricing_context.get("same_bond_secondary_available"),
             "comparable_secondary_market": pricing_context.get("comparable_secondary_market") or [],
             "comparable_median_yield_percent": pricing_context.get("comparable_median_yield_percent"),
@@ -1065,7 +1083,7 @@ def _bond_execution_note(asset: OptimizerAsset) -> str:
         parts.append(
             f"recent primary auction reference {top.get('bond_no')} "
             f"from {top.get('auction_date')} "
-            f"({top.get('maturity_years')}y) coupon/yield proxy "
+            f"({top.get('maturity_years')}y) approximate reinvestment yield "
             f"{_safe_float(top.get('implied_yield_percent')):.2f}% at price "
             f"{_safe_float(top.get('price_per_100')):.2f}"
         )
@@ -1098,6 +1116,138 @@ def _bond_return_worth_note(asset: OptimizerAsset) -> str:
     )
 
 
+def _bond_coupon_advantage(asset: OptimizerAsset) -> float:
+    liquidity = asset.liquidity or {}
+    coupon = _safe_float(liquidity.get("coupon_rate_percent")) / 100
+    reinvestment_yield = _safe_float(liquidity.get("primary_reinvestment_yield_percent")) / 100
+    if coupon <= 0 or reinvestment_yield <= 0:
+        return 0.0
+    return coupon - reinvestment_yield
+
+
+def _is_irreplaceable_income_bond(asset: OptimizerAsset) -> bool:
+    return (
+        asset.asset_type == "BOND"
+        and _bond_coupon_advantage(asset) >= HIGH_COUPON_ADVANTAGE_BPS / 10_000
+    )
+
+
+def _has_rock_solid_fundamentals(asset: OptimizerAsset) -> bool:
+    fundamentals = asset.fundamentals or {}
+    if asset.asset_type != "STOCK":
+        return False
+    pe_ratio = _safe_float(fundamentals.get("pe_ratio"))
+    pb_ratio = _safe_float(fundamentals.get("pb_ratio"))
+    eps = _safe_float(fundamentals.get("eps"))
+    debt_to_equity = _safe_float(fundamentals.get("debt_to_equity"), default=-1)
+    return (
+        eps > 0
+        and 0 < pe_ratio <= 22
+        and 0 < pb_ratio <= 3.5
+        and 0 <= debt_to_equity <= 150
+    )
+
+
+def _has_visible_macro_tailwind(asset: OptimizerAsset) -> bool:
+    data = {**(asset.fundamentals or {}), **(asset.liquidity or {})}
+    return bool(
+        data.get("macro_tailwind")
+        or data.get("sector_tailwind")
+        or data.get("rate_tailwind")
+        or str(data.get("macro_context") or "").lower() in {"tailwind", "supportive", "positive"}
+    )
+
+
+def _clean_exit_liquidity(asset: OptimizerAsset, trade_value: float) -> bool:
+    liquidity = asset.liquidity or {}
+    if asset.asset_type == "STOCK":
+        max_buy_value = _safe_float(liquidity.get("max_buy_value_20d"))
+        return max_buy_value > 0 and trade_value <= max_buy_value
+    if asset.asset_type == "FUND":
+        redemption_days = _safe_float(liquidity.get("redemption_days"), default=999)
+        return redemption_days <= 7 and bool(liquidity.get("prospectus_backed"))
+    return False
+
+
+def _replacement_quality_for_irreplaceable_bond(
+    bond: OptimizerAsset,
+    assets: list[OptimizerAsset],
+    suggested_weights: np.ndarray,
+    current_weights: np.ndarray,
+    total_value: float,
+) -> tuple[bool, list[str]]:
+    coupon_advantage = _bond_coupon_advantage(bond)
+    required_excess = coupon_advantage + LEGACY_BOND_REPLACEMENT_MARGIN
+    increased_assets = [
+        (asset, float(suggested - current) * total_value)
+        for asset, suggested, current in zip(assets, suggested_weights, current_weights)
+        if asset.key != bond.key and suggested > current + 0.01
+    ]
+    failures: list[str] = []
+    for asset, trade_value in increased_assets:
+        net_return_advantage = asset.expected_return - asset.fee_rate - bond.income_yield
+        gates = {
+            "return_margin": net_return_advantage >= required_excess,
+            "low_moderate_volatility": asset.volatility <= LOW_MODERATE_VOLATILITY_MAX,
+            "rock_solid_fundamentals": _has_rock_solid_fundamentals(asset),
+            "macro_tailwind": _has_visible_macro_tailwind(asset),
+            "history_and_liquidity": (
+                _safe_float((asset.liquidity or {}).get("history_years")) >= MIN_REPLACEMENT_HISTORY_YEARS
+                and _clean_exit_liquidity(asset, trade_value)
+            ),
+        }
+        if all(gates.values()):
+            return True, []
+        failed = ", ".join(name for name, passed in gates.items() if not passed)
+        failures.append(f"{asset.symbol} failed replacement gates: {failed}")
+    if not failures:
+        failures.append("No increased replacement asset was available for the income stream.")
+    return False, failures
+
+
+def _protect_irreplaceable_bonds(
+    current_weights: np.ndarray,
+    suggested_weights: np.ndarray,
+    assets: list[OptimizerAsset],
+    total_value: float,
+) -> tuple[np.ndarray, list[str]]:
+    protected = suggested_weights.copy()
+    warnings: list[str] = []
+    for index, asset in enumerate(assets):
+        if not _is_irreplaceable_income_bond(asset) or protected[index] >= current_weights[index]:
+            continue
+        qualified, failures = _replacement_quality_for_irreplaceable_bond(
+            asset,
+            assets,
+            protected,
+            current_weights,
+            total_value,
+        )
+        if qualified:
+            continue
+        restored = float(current_weights[index] - protected[index])
+        protected[index] = current_weights[index]
+        warnings.append(
+            f"{asset.symbol} reduction blocked: coupon materially exceeds recent primary reinvestment yield, so default is hold to maturity unless all replacement gates pass. "
+            + " ".join(failures[:2])
+        )
+        receiver_indices = [
+            i
+            for i, (current, suggested) in enumerate(zip(current_weights, protected))
+            if i != index and suggested > current
+        ]
+        receiver_total = sum(float(protected[i] - current_weights[i]) for i in receiver_indices)
+        if receiver_total > 0:
+            for i in receiver_indices:
+                reduction = restored * float(protected[i] - current_weights[i]) / receiver_total
+                protected[i] = max(current_weights[i], protected[i] - reduction)
+
+    total = float(protected.sum())
+    if total > 0:
+        protected = protected / total
+    return protected, warnings
+
+
 def _optimize_sync(
     assets_payload: list[dict[str, Any]],
     simulations: int,
@@ -1147,6 +1297,13 @@ def _optimize_sync(
         total_value,
         settings,
     )
+    best_weights, bond_protection_warnings = _protect_irreplaceable_bonds(
+        current_weights,
+        best_weights,
+        assets,
+        total_value,
+    )
+    liquidity_warnings.extend(bond_protection_warnings)
     liquidity_warnings.extend(
         _liquidity_trade_warnings(current_weights, best_weights, assets, total_value)
     )
@@ -1177,6 +1334,11 @@ def _optimize_sync(
             if asset.asset_type == "BOND":
                 reason = f"{reason}. {_bond_return_worth_note(asset)} {_bond_execution_note(asset)}"
         elif asset.asset_type == "BOND":
+            if _is_irreplaceable_income_bond(asset):
+                reason = (
+                    "Hold to maturity by default: coupon materially exceeds recent primary reinvestment yields, "
+                    "and no replacement passed all return, volatility, fundamentals, macro, history, and liquidity gates"
+                )
             reason = f"{reason}. {_bond_return_worth_note(asset)} {_bond_execution_note(asset)}"
         allocations.append(
             {

App/routers/economy/valuation.py

@@ -92,13 +92,22 @@ async def get_comparable_primary_bonds(
     target_coupon = float(getattr(bond, "coupon_rate", 0) or 0)
     rows = await Bond.filter(auction_date__gte=min_auction_date).order_by("-auction_date").limit(500)
 
-    def score(row: Bond) -> tuple[float, int, float, int]:
+    def maturity_years(row: Bond) -> float:
         try:
-            years = float(str(row.maturity_years or "").split()[0])
+            return float(str(row.maturity_years or "").split()[0])
         except Exception:
-            years = target_years if target_years is not None else 0
+            return target_years if target_years is not None else 0
+
+    def approximate_yield(row: Bond) -> float:
+        years = max(maturity_years(row), 1.0)
+        coupon = float(row.coupon_rate or 0)
+        price = float(row.price_per_100 or 100)
+        return max(((coupon + ((100 - price) / years)) / ((100 + price) / 2)) * 100, 0.0)
+
+    def score(row: Bond) -> tuple[float, int, float, int]:
+        years = maturity_years(row)
         years_gap = abs((years or 0) - (target_years or years or 0))
-        coupon_gap = abs(float(row.coupon_rate or 0) - target_coupon)
+        coupon_gap = abs(approximate_yield(row) - target_coupon)
         auction_recency = -int(row.auction_date.toordinal()) if row.auction_date else 0
         return (years_gap, auction_recency, coupon_gap, -int(row.auction_number or 0))
 
@@ -113,7 +122,7 @@ async def get_comparable_primary_bonds(
             "maturity_date": row.maturity_date.isoformat(),
             "coupon_rate": row.coupon_rate,
             "price_per_100": row.price_per_100,
-            "implied_yield_percent": row.coupon_rate,
+            "implied_yield_percent": round(approximate_yield(row), 4),
             "source": "primary_auction_reference",
         }
         for row in sorted(comparable, key=score)[:limit]

tests/test_portfolio_optimizer_fundamentals.py

@@ -7,6 +7,7 @@ from App.analysis.portfolio_optimizer import (
     _bond_return_worth_note,
     _fundamental_adjusted_stock_return,
     _liquidity_capacity_penalty,
+    _protect_irreplaceable_bonds,
 )
 from App.routers.stocks.fundamentals import parse_market_table, parse_stock_page
 
@@ -140,3 +141,54 @@ def test_bond_worth_note_uses_market_yield_not_coupon():
     assert "market yield 9.03%" in note
     assert "15.50%" not in note
     assert "coupon income is not the same as total return" in note
+
+
+def test_irreplaceable_bond_reduction_is_blocked_without_all_replacement_gates():
+    current = np.array([0.5, 0.5])
+    suggested = np.array([0.2, 0.8])
+    bond = OptimizerAsset(
+        key="BOND:1",
+        asset_id=1,
+        asset_type="BOND",
+        symbol="TZ1996103283",
+        name="Legacy Treasury Bond",
+        quantity=1,
+        current_price=1.5,
+        current_value=500_000,
+        expected_return=0.09,
+        volatility=0.04,
+        income_yield=0.103,
+        fee_rate=0.0,
+        liquidity={
+            "coupon_rate_percent": 15.5,
+            "primary_reinvestment_yield_percent": 9.0,
+            "price_percent": 150,
+        },
+    )
+    stock = OptimizerAsset(
+        key="STOCK:1",
+        asset_id=1,
+        asset_type="STOCK",
+        symbol="SPEC",
+        name="Speculative Stock",
+        quantity=1,
+        current_price=1,
+        current_value=500_000,
+        expected_return=0.3,
+        volatility=0.28,
+        income_yield=0.0,
+        fee_rate=0.0206,
+        fundamentals={"eps": 10, "pe_ratio": 12, "pb_ratio": 2, "debt_to_equity": 50},
+        liquidity={"history_years": 4, "max_buy_value_20d": 1_000_000},
+    )
+
+    protected, warnings = _protect_irreplaceable_bonds(
+        current,
+        suggested,
+        [bond, stock],
+        1_000_000,
+    )
+
+    assert protected[0] >= current[0]
+    assert protected[1] <= current[1]
+    assert warnings