Signal Algorithm Optimization Plan (v4.2 Implementation Blueprint)

現況摘要

系統已有完整的 Signal 基礎架構： - api/enums.py: SignalChannel, Bias（無 RiskFlag） - api/routes/signals.py: 3 channel 的判定邏輯（Momentum / Reversal / Squeeze） - scripts/daily_analysis_snapshot.py: compute_bias_confidence_reasons_warnings() 計算 confidence score - tests/test_signals_snapshot_rules.py: 現有單元測試

---

要解決的 4 個核心問題

問題 1：RSI 懲罰分流 (當前 Bug)

現況：compute_bias_confidence_reasons_warnings() line 461：

elif rsi_value is not None and rsi_value <= 30:
    score += 0.03  # 超賣給正分 → 但 REVERSAL_BULLISH 在臨界點仍可能被扣

問題：RSI_OVERBOUGHT (-0.05) 與 RSI_OVERSOLD (+0.03) 是對稱問題，但語義不同—— 超賣是抄底訊號的有利條件，不應被視為 penalty 來「抵消」，而應在 Reversal context 完全免懲罰。

問題 2：Breakout 兩段式失效 (當前缺漏)

現況：_passes_squeeze() 只用 bb_status in ("contracting", "squeezing") 判定， 沒有處理「已突破但尚未轉 Momentum」的過渡狀態，導致突破訊號直接消失。

問題 3：無 RiskFlag 結構化記錄

現況：warnings 是 List[str]，無法做結構化查詢、penalty 計算或 diff report。

問題 4：Confidence 累加無結構化 Penalty Table

現況：分散的 score += / -= 邏輯，難以審計、擴充或針對不同 channel 調整。

---

實作範圍（最小改動原則）

不動資料庫 schema，不改 API contract，只修 Python logic。

Step 1：enums.py — 新增 RiskFlag, QualityFlag

檔案：api/enums.py

新增：

class RiskFlag(str, Enum):
    RSI_OVERBOUGHT = "rsi_overbought"        # RSI >= 70，僅懲罰多方趨勢
    RSI_OVERSOLD = "rsi_oversold"            # RSI <= 30，不懲罰，僅記錄
    MACD_DIVERGENCE = "macd_divergence"      # MACD 背離
    RSI_DIVERGENCE_BEARISH = "rsi_divergence_bearish"   # RSI 空方背離警告
    RSI_DIVERGENCE_BULLISH = "rsi_divergence_bullish"   # RSI 多方背離警告
    BB_EXPANDING = "bb_expanding"            # BB 頻寬擴張中
    BB_BREAKOUT_WATCH = "bb_breakout_watch"  # 弱突破觀察期
    VOLUME_DIVERGENCE = "volume_divergence"  # 量能背離

class QualityFlag(str, Enum):
    COMPLETE = "complete"
    PARTIAL = "partial"
    LOW_LIQUIDITY = "low_liquidity"

---

Step 2：daily_analysis_snapshot.py — RSI 分流修正

檔案：scripts/daily_analysis_snapshot.py

修改目標：compute_bias_confidence_reasons_warnings() (line 457-469)

現況邏輯：

if rsi_value >= 70:
    score -= 0.05      # 超買 penalty
elif rsi_value <= 30:
    score += 0.03      # 超賣 bonus（雙向對稱，語義不清）

改後邏輯：

if rsi_value >= 70:
    warnings.append(RiskFlag.RSI_OVERBOUGHT)
    # 僅在 bias 為多方時懲罰（趨勢過熱）
    if current_bias_direction != Bias.BEARISH:
        score -= 0.05
elif rsi_value <= 30:
    warnings.append(RiskFlag.RSI_OVERSOLD)
    # 超賣：不列入 penalty，僅記錄（Reversal Bullish 的有利條件）
    # score 不動

注意：compute_bias_confidence_reasons_warnings() 在計算 score 時尚未知道最終 bias， 所以用「中性判斷」：只要 score 當前趨勢為正（> 0.5）且 RSI 超買，才懲罰。

實際改法更簡單：

# 超買懲罰：只在 score 目前偏多方才扣（避免空方訊號被誤傷）
if rsi_value >= 70:
    warnings_list.append(RiskFlag.RSI_OVERBOUGHT)
    if score > 0.5:          # 當前趨勢偏多，才算過熱風險
        score -= 0.05
# 超賣：純記錄，不懲罰（不是多方的風險，是機會）
elif rsi_value <= 30:
    warnings_list.append(RiskFlag.RSI_OVERSOLD)
    # score 不變（移除原來的 +0.03）

---

Step 3：signals.py — Breakout 兩段式緩衝

檔案：api/routes/signals.py

修改目標：_passes_squeeze() 函數 + Squeeze 評分段落

新增常數：

SQUEEZE_BW_EXPLODE_RATIO = 1.3   # BB 頻寬擴張門檻

新增輔助函數：

def _is_price_breakout(row: Dict[str, Any], direction: Bias) -> bool:
    """價格是否已突破 BB 軌道。"""
    close = _to_float(row.get("close"))
    bb_upper = _to_float(row.get("bb_upper"))
    bb_lower = _to_float(row.get("bb_lower"))
    if close is None:
        return False
    if direction == Bias.BULLISH and bb_upper is not None:
        return close > bb_upper
    if direction == Bias.BEARISH and bb_lower is not None:
        return close < bb_lower
    return False


def _is_bw_exploding(row: Dict[str, Any], prev_row: Optional[Dict[str, Any]]) -> bool:
    """BB 頻寬是否開始爆發擴張。"""
    bw = _to_float(row.get("bb_bandwidth"))
    prev_bw = _to_float(prev_row.get("bb_bandwidth")) if prev_row else None
    if bw is None or prev_bw is None or prev_bw == 0:
        return False
    return bw / prev_bw >= SQUEEZE_BW_EXPLODE_RATIO

修改 Squeeze 判定邏輯（在迴圈內）：

# 現況：只有 _passes_squeeze() 通過才進 squeeze
# 改後：加入 Breakout Watch 狀態

if _passes_squeeze(row_dict, Bias.BULLISH):
    price_bo = _is_price_breakout(row_dict, Bias.BULLISH)
    bw_explode = _is_bw_exploding(row_dict, None)  # prev_row 需從 query 取

    if price_bo and bw_explode:
        # 強 Breakout → 不進 squeeze，讓 momentum 接棒（不輸出 squeeze 訊號）
        pass
    elif price_bo and not bw_explode:
        # 弱 Breakout → 進入觀察期，仍輸出 squeeze 但加 BREAKOUT_WATCH flag
        desc = _build_description(row_dict, "squeeze", 0, "breakout_watch=yes")
        signal = _build_signal_response(...)
        signal_with_flag = signal  # 可在 description 或新欄位標記
        squeeze_bull.append((sort_key, signal))
    else:
        # 正常 squeeze
        desc = _build_description(row_dict, "squeeze", 0)
        signal = _build_signal_response(...)
        squeeze_bull.append((sort_key, signal))

Query 補充：需在 SQL 查詢中加入 bb_upper, bb_lower, bb_bandwidth 以及前一日的 prev_bb_bandwidth（LAG）：

r.bb_upper,
r.bb_lower,
LAG(r.bb_bandwidth) OVER (PARTITION BY r.ticker ORDER BY r.trading_date) AS prev_bb_bandwidth

---

Step 4：signals.py — Confidence Penalty 結構化 (可選強化)

這步驟是讓 Penalty 可審計，但不改變數值結果。目前 signals.py 中直接用 confidence 值做排序， 若要加 Diff Report，可在 _build_description() 中輸出 penalty breakdown。

現況 description：

MA alignment bullish / ADX trending | trend_hits=4 | macd_expand_2d=yes

改後 description 加入 penalty 資訊：

MA alignment bullish / ADX trending | trend_hits=4 | macd_expand_2d=yes | penalties=RSI_OVERBOUGHT(-0.05)

修改 _build_description() 接受 risk_flags: List[str] 參數，附加到輸出末尾。

---

Step 5：測試補充

檔案：tests/test_signals_snapshot_rules.py

新增測試案例：

1. test_rsi_oversold_no_penalty_for_reversal_bullish
2. RSI = 28，bias = bullish，score 不應因超賣被扣分

3. compute_bias_confidence_reasons_warnings() 輸出 confidence 應 >= 無 RSI penalty 版本

4. test_rsi_overbought_penalty_only_when_bullish

5. RSI = 72，score > 0.5 → penalty -0.05

6. RSI = 72，score < 0.5（空方趨勢）→ 不扣分

7. test_squeeze_strong_breakout_not_emitted

8. bb_status=squeezing + price > bb_upper + bw exploding → 不輸出 squeeze 訊號

9. test_squeeze_weak_breakout_emitted_with_watch_flag

10. bb_status=squeezing + price > bb_upper + bw NOT exploding → 輸出 squeeze，description 含 "breakout_watch=yes"

---

修改檔案清單

檔案	修改內容	影響範圍
api/enums.py	新增 RiskFlag, QualityFlag	全系統 import
scripts/daily_analysis_snapshot.py	RSI penalty 分流	Confidence score 計算
api/routes/signals.py	Breakout 兩段式 + bb 欄位補入 SQL + description penalty	Signal 輸出
tests/test_signals_snapshot_rules.py	新增 4 個測試案例	測試覆蓋率

不需改動： - DB schema（不新增欄位） - API response contract（SignalResponse 不變） - build_agent_payload() 或 snapshot 儲存邏輯

---

執行順序

1. api/enums.py → 新增 enum（其他步驟依賴此）
2. scripts/daily_analysis_snapshot.py → RSI penalty 修正
3. api/routes/signals.py → SQL 補欄位 + Breakout 兩段式
4. tests/test_signals_snapshot_rules.py → 測試補充
5. 跑測試驗證