Merge branch 'main' of https://github.com/handsomejustin/easy_tdx

src/easy_tdx/__init__.py

@@ -27,6 +27,7 @@ from .ex.mac_client import AsyncMacExClient, MacExClient
 from .ex.models import KNOWN_EX_HOSTS
 from .exceptions import TdxCommandError, TdxConnectionError, TdxDecodeError, TdxError
 from .mac.client import AsyncMacClient, MacClient
+from .mac.commands import UNUSUAL_TYPE_NAMES
 from .mac.enums import (
     Adjust,
     BoardSortColumn,
@@ -90,6 +91,7 @@ __all__ = [
     "TransactionRecord",
     "XdxrRecord",
     "XDXR_CATEGORY_NAMES",
+    "UNUSUAL_TYPE_NAMES",
     "FinanceInfo",
     "CompanyInfoCategory",
     "FinancialFileInfo",

src/easy_tdx/backtest/benchmark.py

@@ -15,7 +15,9 @@
       "fitness": {"pass_ratio": 0.875, "high_fitness": true, "checks": [...]},
       "benchmark": {
         "buy_hold": {"total_return": 0.32, ...},
-        "excess_return": 0.18                               # 策略 - 买入持有
+        "excess_return": 0.18,                             # 策略 - 买入持有
+        "alpha": 0.09, "beta": 0.72,                       # v1.28：CAPM 对比
+        "information_ratio": 0.85, "tracking_error": 0.12  # v1.28：主动管理指标
       },
       "config": {...}
     }
@@ -27,8 +29,9 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
+import numpy as np
 import pandas as pd
 
 from easy_tdx.backtest.engine import BacktestEngine
@@ -39,7 +42,16 @@ from easy_tdx.backtest.strategy import Strategy
 from easy_tdx.backtest.types import to_json_native
 from easy_tdx.backtest.walkforward import WalkForwardEngine
 
-__all__ = ["evaluate_strategy", "run_buy_hold_benchmark"]
+if TYPE_CHECKING:
+    import numpy.typing as npt
+
+    from easy_tdx.backtest.types import BacktestResult
+
+    NDArray = npt.NDArray[np.float64]
+else:
+    NDArray = np.ndarray
+
+__all__ = ["evaluate_strategy", "run_buy_hold_benchmark", "compute_benchmark_comparison"]
 
 
 class _BuyAndHold(Strategy):
@@ -54,6 +66,32 @@ class _BuyAndHold(Strategy):
             self._bought = True
 
 
+def _run_buy_hold_result(
+    df: pd.DataFrame,
+    cash: float = 100000.0,
+    commission: float = 0.0003,
+    min_commission: float = 5.0,
+    stamp_tax: float = 0.001,
+    slippage: float = 0.0,
+    execution: str = "next_open",
+    symbol: str | None = None,
+    auto_fees: bool = False,
+) -> BacktestResult:
+    """买入持有基准完整回测（内部用，返回 BacktestResult 以取资金曲线）。"""
+    engine = BacktestEngine(
+        strategy=_BuyAndHold,
+        cash=cash,
+        commission=commission,
+        min_commission=min_commission,
+        stamp_tax=stamp_tax,
+        slippage=slippage,
+        execution=execution,
+        symbol=symbol,
+        auto_fees=auto_fees,
+    )
+    return engine.run(df)
+
+
 def run_buy_hold_benchmark(
     df: pd.DataFrame,
     cash: float = 100000.0,
@@ -66,18 +104,9 @@ def run_buy_hold_benchmark(
     auto_fees: bool = False,
 ) -> dict[str, Any]:
     """买入持有基准回测（与策略回测同区间、同费率、同资金）。"""
-    engine = BacktestEngine(
-        strategy=_BuyAndHold,
-        cash=cash,
-        commission=commission,
-        min_commission=min_commission,
-        stamp_tax=stamp_tax,
-        slippage=slippage,
-        execution=execution,
-        symbol=symbol,
-        auto_fees=auto_fees,
+    result = _run_buy_hold_result(
+        df, cash, commission, min_commission, stamp_tax, slippage, execution, symbol, auto_fees
     )
-    result = engine.run(df)
     keys = (
         "total_return",
         "annual_return",
@@ -89,6 +118,77 @@ def run_buy_hold_benchmark(
     return dict(to_json_native({k: result.performance.get(k, 0.0) for k in keys}))
 
 
+def compute_benchmark_comparison(
+    strategy_curve: pd.DataFrame,
+    benchmark_curve: pd.DataFrame,
+    annual_days: int = 252,
+) -> dict[str, float]:
+    """策略 vs 基准的 CAPM / 主动管理对比指标（v1.28 新增）。
+
+    从两条资金曲线的日收益率序列计算：
+
+    - ``beta``: 协方差/基准方差，策略对基准的敏感度（1 = 与基准同涨跌）
+    - ``alpha``: 年化 CAPM α ≈ (策略日均收益 − β×基准日均收益) × 年化天数，
+      简化版（无风险利率并入截距），>0 说明剔除基准影响后仍有超额
+    - ``information_ratio``: 年化信息比率 = mean(策略−基准)/std(策略−基准)×√N，
+      每 1 单位跟踪误差换来多少超额收益
+    - ``tracking_error``: 年化跟踪误差 = std(策略−基准)×√N
+
+    两条曲线按 bar 对齐（截取较短长度）；基准方差为 0（曲线恒定）时
+    beta/alpha 记 0，IR 在差值恒正且无波动时沿用 999 上限约定。
+
+    Args:
+        strategy_curve: 策略资金曲线（含 total 列）
+        benchmark_curve: 基准资金曲线（含 total 列）
+        annual_days: 年化交易日数
+
+    Returns:
+        {alpha, beta, information_ratio, tracking_error}
+    """
+    s_total = strategy_curve["total"].to_numpy(dtype=np.float64)
+    b_total = benchmark_curve["total"].to_numpy(dtype=np.float64)
+    n = min(len(s_total), len(b_total))
+    if n < 3:
+        return {"alpha": 0.0, "beta": 0.0, "information_ratio": 0.0, "tracking_error": 0.0}
+
+    def _daily_ret(total: NDArray) -> NDArray:
+        safe_prev = np.where(total[:-1] != 0, total[:-1], np.nan)
+        ret = np.diff(total) / safe_prev
+        return ret[np.isfinite(ret)]
+
+    s_ret = _daily_ret(s_total[:n])
+    b_ret = _daily_ret(b_total[:n])
+    m = min(len(s_ret), len(b_ret))
+    if m < 2:
+        return {"alpha": 0.0, "beta": 0.0, "information_ratio": 0.0, "tracking_error": 0.0}
+    s_ret, b_ret = s_ret[:m], b_ret[:m]
+
+    b_var = float(np.var(b_ret))
+    if b_var > 1e-18:
+        beta = float(np.cov(s_ret, b_ret)[0, 1] / b_var)
+        alpha = float((np.mean(s_ret) - beta * np.mean(b_ret)) * annual_days)
+    else:
+        beta = 0.0
+        alpha = float(np.mean(s_ret) * annual_days)
+
+    diff = s_ret - b_ret
+    diff_std = float(np.std(diff))
+    if diff_std > 1e-12:
+        information_ratio = float(np.mean(diff) / diff_std * np.sqrt(annual_days))
+    elif np.mean(diff) > 0:
+        information_ratio = 999.0
+    else:
+        information_ratio = 0.0
+    tracking_error = diff_std * np.sqrt(annual_days)
+
+    return {
+        "alpha": alpha,
+        "beta": beta,
+        "information_ratio": information_ratio,
+        "tracking_error": tracking_error,
+    }
+
+
 def evaluate_strategy(
     strategy: type[Strategy] | Strategy,
     df: pd.DataFrame,
@@ -153,8 +253,11 @@ def evaluate_strategy(
     score = score_strategy(perf, wf=wf)
     grade = grade_performance(perf)
 
-    # 5. 基准对比（买入持有，同区间同费率）
-    bh = run_buy_hold_benchmark(df, **engine_kwargs)
+    # 5. 基准对比（买入持有，同区间同费率）：超额收益 + Alpha/Beta/IR/TE
+    bh_result = _run_buy_hold_result(df, **engine_kwargs)
+    bh_keys = ("total_return", "annual_return", "max_drawdown", "sharpe", "calmar", "volatility")
+    bh = dict(to_json_native({k: bh_result.performance.get(k, 0.0) for k in bh_keys}))
+    comparison = compute_benchmark_comparison(bt.equity_curve, bh_result.equity_curve)
 
     return {
         "performance": to_json_native(dict(perf)),
@@ -166,6 +269,7 @@ def evaluate_strategy(
             "buy_hold": bh,
             "excess_return": float(perf.get("total_return", 0.0))
             - float(bh.get("total_return", 0.0)),
+            **comparison,
         },
         "config": {
             "symbol": symbol,

src/easy_tdx/backtest/cli.py

@@ -318,6 +318,16 @@ def _print_table(result: Any) -> None:
     click.echo(f"夏普比率: {perf.get('sharpe', 0):.2f}")
     click.echo(f"胜率: {perf.get('win_rate', 0):.2%}")
     click.echo(f"交易次数: {perf.get('total_trades', 0)}")
+    # 深度风险指标（v1.28 新增；老结果缺键时跳过，不输出 0 假值）
+    if perf.get("ulcer_index") is not None:
+        click.echo(f"Ulcer 指数: {perf.get('ulcer_index', 0):.4f}")
+        click.echo(f"日 VaR(95%): {perf.get('var_95', 0):.2%}")
+        click.echo(f"日 CVaR(95%): {perf.get('cvar_95', 0):.2%}")
+        click.echo(f"SQN 系统质量: {perf.get('sqn', 0):.2f}")
+        click.echo(
+            f"最大连胜/连亏: {perf.get('max_consecutive_wins', 0)} / "
+            f"{perf.get('max_consecutive_losses', 0)}"
+        )
     click.echo()
 
     if getattr(result, "diagnostic", None):

src/easy_tdx/backtest/engine.py

@@ -37,13 +37,32 @@ if TYPE_CHECKING:
 class _StopCondition:
     """Active stop-loss / take-profit condition tied to an open position.
 
+    三条退出线构成 OCO：任一触发即整体失效（见 ``_check_stop_conditions``）。
+
     Attributes:
         stop_loss: Price below which a SELL is triggered (None = disabled)
         take_profit: Price above which a SELL is triggered (None = disabled)
+        trail_stop: Trailing stop percent (e.g. 0.08 = 8% below the highest
+            close since entry, None = disabled). Fixed ``stop_loss`` wins when
+            both are set.
+        high_watermark: Highest close seen since the BUY (trailing reference).
+            Updated at the END of each bar (after the trigger check), so a
+            trailing stop can only fire from the NEXT bar onward — consistent
+            with next_open execution semantics.
     """
 
     stop_loss: float | None
     take_profit: float | None
+    trail_stop: float | None = None
+    high_watermark: float = 0.0
+
+    def effective_stop(self) -> float | None:
+        """当前生效的止损价（固定价优先，其次移动止损；均无则 None）。"""
+        if self.stop_loss is not None:
+            return self.stop_loss
+        if self.trail_stop is not None:
+            return self.high_watermark * (1.0 - self.trail_stop)
+        return None
 
 
 class BacktestEngine:
@@ -375,10 +394,17 @@ class BacktestEngine:
             # activate on the NEXT bar — consistent with next_open execution)
             for sig in bar_signals:
                 if sig.direction == "BUY" and (
-                    sig.stop_loss is not None or sig.take_profit is not None
+                    sig.stop_loss is not None
+                    or sig.take_profit is not None
+                    or sig.trail_stop is not None
                 ):
                     active_stops.append(
-                        _StopCondition(stop_loss=sig.stop_loss, take_profit=sig.take_profit)
+                        _StopCondition(
+                            stop_loss=sig.stop_loss,
+                            take_profit=sig.take_profit,
+                            trail_stop=sig.trail_stop,
+                            high_watermark=close_arr[i],
+                        )
                     )
 
             # Clear conditions when a SELL occurs (strategy or SL/TP triggered)
@@ -488,8 +514,11 @@ class BacktestEngine:
         """Check active SL/TP conditions against current bar's price range.
 
         If triggered, generates a SELL signal at the trigger price and removes
-        the condition. Stop-loss is checked first (conservative: assume the
-        worst case for the holder).
+        the condition (OCO: all remaining legs of the same condition die too).
+        Stop-loss is checked first (conservative: assume the worst case for
+        the holder). Trailing stops reference the highest close seen through
+        the PREVIOUS bar (watermark is updated after the check), so they can
+        never fire on the same bar that sets a new high.
 
         Args:
             active_stops: List of active stop conditions
@@ -512,16 +541,22 @@ class BacktestEngine:
             triggered = False
             trigger_price = 0.0
 
-            # Check stop-loss first (worst case for holder)
-            if cond.stop_loss is not None and bar_low <= cond.stop_loss:
+            # Check stop-loss first (worst case for holder); trailing resolves
+            # to its effective price, fixed stop_loss wins if both set
+            eff_stop = cond.effective_stop()
+            if eff_stop is not None and bar_low <= eff_stop:
                 triggered = True
-                trigger_price = cond.stop_loss
+                trigger_price = eff_stop
             # Then check take-profit
             elif cond.take_profit is not None and bar_high >= cond.take_profit:
                 triggered = True
                 trigger_price = cond.take_profit
 
-            if triggered:
+            if not triggered:
+                # Trailing watermark update AFTER the check (close-based)
+                cond.high_watermark = max(cond.high_watermark, bar_close)
+                remaining.append(cond)
+            else:
                 # Get datetime for this bar
                 dt_val = df["datetime"].iloc[bar_index]
                 if hasattr(dt_val, "strftime"):
@@ -538,8 +573,6 @@ class BacktestEngine:
                         source="stop",  # 标记为止损/止盈触发，延迟到下一根成交
                     )
                 )
-            else:
-                remaining.append(cond)
 
         active_stops.clear()
         active_stops.extend(remaining)

src/easy_tdx/backtest/performance.py

@@ -22,7 +22,8 @@ else:
 class PerformanceAnalyzer:
     """绩效分析器。
 
-    从资金曲线和交易记录计算 19 项绩效指标。
+    从资金曲线和交易记录计算 25 项绩效指标（19 项经典指标 + 6 项
+    深度风险指标：Ulcer / VaR / CVaR / SQN / 最大连胜连亏，v1.28 新增）。
 
     Attributes:
         ANNUAL_DAYS: 年化交易日数（默认 252）
@@ -55,7 +56,7 @@ class PerformanceAnalyzer:
         """计算绩效指标。
 
         Returns:
-            包含 19 项指标的字典：
+            包含 25 项指标的字典：
             - total_return: 总收益率
             - annual_return: 年化收益率
             - max_drawdown: 最大回撤
@@ -75,6 +76,14 @@ class PerformanceAnalyzer:
             - max_loss: 最大亏损
             - avg_holding_days: 平均持仓天数（FIFO 配对、按 size 加权，日历日口径）
             - volatility: 年化波动率
+            - ulcer_index: Ulcer 指数（回撤深度平方均值的开方，综合反映
+              回撤深度与持续时间，越小持有体验越好）
+            - var_95: 95% 日 VaR（历史分位数法，正数表示单日最大损失幅度）
+            - cvar_95: 95% 日 CVaR / 期望损失（尾部 5% 日收益均值，正数）
+            - sqn: 系统质量数（Van Tharp SQN = √N × 单笔收益率均值/标准差，
+              >2 可用、>4 优秀、>6 极佳的经验分档）
+            - max_consecutive_wins: 最大连胜笔数（按 SELL 成交顺序统计）
+            - max_consecutive_losses: 最大连亏笔数
         """
         # 边界检查
         if len(self._equity_curve) < 2:
@@ -211,6 +220,28 @@ class PerformanceAnalyzer:
         # 19. 年化波动率
         volatility = np.std(daily_ret) * np.sqrt(self.ANNUAL_DAYS)
 
+        # 20. Ulcer 指数（Martin：√(mean(回撤幅度²))，深度与持续时间加权）
+        ulcer_index = float(np.sqrt(np.mean(drawdown_pct**2)))
+
+        # 21. 95% 日 VaR（历史分位数法；正数表示损失幅度，便于直觉解读）
+        var_95 = float(-np.percentile(daily_ret, 5))
+
+        # 22. 95% 日 CVaR（VaR 之外尾部收益的均值；样本不足时退化为 VaR）
+        tail = daily_ret[daily_ret <= -var_95]
+        cvar_95 = float(-np.mean(tail)) if len(tail) > 0 else var_95
+
+        # 23. SQN 系统质量数（√N × 单笔收益率均值 / 标准差）
+        valid_tr = trade_returns[np.isfinite(trade_returns)]
+        if len(valid_tr) >= 2 and np.std(valid_tr) > 1e-12:
+            sqn = float(np.sqrt(len(valid_tr)) * np.mean(valid_tr) / np.std(valid_tr))
+        else:
+            sqn = 0.0
+
+        # 24/25. 最大连胜/连亏（与 win_rate 同口径：按 SELL 成交顺序）
+        max_consecutive_wins, max_consecutive_losses = self._max_win_lose_streaks(
+            sell_trades["pnl"].to_numpy(dtype=np.float64)
+        )
+
         return {
             "total_return": total_return,
             "annual_return": annual_return,
@@ -231,6 +262,12 @@ class PerformanceAnalyzer:
             "max_loss": max_loss,
             "avg_holding_days": avg_holding_days,
             "volatility": volatility,
+            "ulcer_index": ulcer_index,
+            "var_95": var_95,
+            "cvar_95": cvar_95,
+            "sqn": sqn,
+            "max_consecutive_wins": max_consecutive_wins,
+            "max_consecutive_losses": max_consecutive_losses,
             # 别名键（兼容常见叫法，避免 .get('sharpe_ratio') 等误用返回 0）
             "sharpe_ratio": sharpe,
             "start_cash": float(total[0]),
@@ -371,7 +408,37 @@ class PerformanceAnalyzer:
             "max_loss": 0.0,
             "avg_holding_days": 0.0,
             "volatility": 0.0,
+            "ulcer_index": 0.0,
+            "var_95": 0.0,
+            "cvar_95": 0.0,
+            "sqn": 0.0,
+            "max_consecutive_wins": 0,
+            "max_consecutive_losses": 0,
             "sharpe_ratio": 0.0,
             "start_cash": 0.0,
             "end_value": 0.0,
         }
+
+    @staticmethod
+    def _max_win_lose_streaks(pnl_seq: NDArray) -> tuple[int, int]:
+        """按成交顺序统计最大连胜/连亏笔数。
+
+        pnl > 0 记为胜，pnl <= 0 记为负（与 win_rate 的胜/负口径一致）。
+
+        Args:
+            pnl_seq: SELL 成交的 pnl 序列（时间升序）
+
+        Returns:
+            (最大连胜笔数, 最大连亏笔数)
+        """
+        max_wins = max_losses = cur_wins = cur_losses = 0
+        for pnl in pnl_seq:
+            if pnl > 0:
+                cur_wins += 1
+                cur_losses = 0
+                max_wins = max(max_wins, cur_wins)
+            else:
+                cur_losses += 1
+                cur_wins = 0
+                max_losses = max(max_losses, cur_losses)
+        return int(max_wins), int(max_losses)

src/easy_tdx/backtest/strategy.py

@@ -336,20 +336,44 @@ class Strategy(ABC):
         price: float | None = None,
         stop_loss: float | None = None,
         take_profit: float | None = None,
+        trail_stop: float | None = None,
+        stop_loss_pct: float | None = None,
+        take_profit_pct: float | None = None,
     ) -> None:
-        """生成买入信号。
+        """生成买入信号（可携带 bracket 止损/止盈/移动止损，OCO 联动）。
+
+        akquant ``place_bracket`` 风格：进出场一体化，不必再手写止损监控。
+        三条退出线任一触发即全部失效（OCO），由引擎逐 bar 监控并自动
+        生成 SELL（``source="stop"``，延迟到下一根开盘成交，消除前视偏差）。
 
         Args:
             size: 交易数量（0 = 全仓，由引擎计算）
             price: 限价（None = 市价单）
-            stop_loss: 止损价（None = 不设置）
-            take_profit: 止盈价（None = 不设置）
+            stop_loss: 止损价（绝对价；与 stop_loss_pct 同时给时绝对价优先）
+            take_profit: 止盈价（绝对价；与 take_profit_pct 同时给时绝对价优先）
+            trail_stop: 移动止损百分比（如 0.08 = 自持仓期间最高收盘价
+                回撤 8% 触发）。固定 stop_loss 优先于移动止损。
+            stop_loss_pct: 止损百分比（相对当前收盘价，如 0.05 = 跌 5% 止损）
+            take_profit_pct: 止盈百分比（相对当前收盘价，如 0.10 = 涨 10% 止盈）
+
+        Examples:
+            >>> # 买入并带 5% 止损 / 10% 止盈（自动换算价格）
+            ... self.buy(stop_loss_pct=0.05, take_profit_pct=0.10)
+            >>> # 买入并带 8% 移动止损（涨得越多止损线跟得越高）
+            ... self.buy(trail_stop=0.08)
         """
         if self._data_proxy is None:
             raise RuntimeError("策略未绑定数据，请先调用 _bind_data()")
         if self._datetime_array is None:
             raise RuntimeError("数据未正确初始化")
 
+        # 百分比便捷参数 → 绝对价（显式绝对价优先）
+        ref_price = price if price is not None else float(self.data.close[0])
+        if stop_loss is None and stop_loss_pct is not None:
+            stop_loss = ref_price * (1.0 - stop_loss_pct)
+        if take_profit is None and take_profit_pct is not None:
+            take_profit = ref_price * (1.0 + take_profit_pct)
+
         signal = Signal(
             datetime=int(self._datetime_array[self._bar_index]),
             direction="BUY",
@@ -357,6 +381,7 @@ class Strategy(ABC):
             price=price,
             stop_loss=stop_loss,
             take_profit=take_profit,
+            trail_stop=trail_stop,
         )
         self._signals.append(signal)
 

src/easy_tdx/backtest/types.py

@@ -25,9 +25,14 @@ class Signal:
         price: 限价（None = 市价单）
         stop_loss: 止损价（None = 不设置）
         take_profit: 止盈价（None = 不设置）
+        trail_stop: 移动止损百分比（如 0.08 = 自持仓期间最高收盘价回撤
+            8% 触发，None = 不设置）。与 ``stop_loss`` 同时设置时固定价优先。
         source: 信号来源。"strategy"=策略产生（默认）；
             "stop"=止损/止盈触发。stop 来源的信号不在信号 bar 当根成交，
             而是延迟到下一根开盘（消除用当根 intrabar 触发价成交的前视偏差）。
+
+    ``stop_loss`` / ``take_profit`` / ``trail_stop`` 三者构成 OCO
+    （one-cancels-other）：任一触发即全部失效，由引擎逐 bar 监控。
     """
 
     datetime: int
@@ -36,6 +41,7 @@ class Signal:
     price: float | None = None
     stop_loss: float | None = None
     take_profit: float | None = None
+    trail_stop: float | None = None
     source: str = "strategy"
 
 

src/easy_tdx/mac/commands/__init__.py

@@ -13,7 +13,7 @@ from .symbol_quotes import SymbolQuotesCmd
 from .symbol_tick_chart import SymbolTickChartCmd
 from .symbol_transaction import SymbolTransactionCmd
 from .tick_charts import TickChartsCmd
-from .unusual import UnusualCmd
+from .unusual import UNUSUAL_TYPE_NAMES, UnusualCmd
 
 __all__ = [
     "BoardListCmd",
@@ -29,5 +29,6 @@ __all__ = [
     "SymbolTickChartCmd",
     "SymbolTransactionCmd",
     "TickChartsCmd",
+    "UNUSUAL_TYPE_NAMES",
     "UnusualCmd",
 ]

src/easy_tdx/mac/commands/unusual.py

@@ -8,9 +8,34 @@ from ...codec.mac_frame import build_mac_request
 from ...commands.base import BaseCommand
 from ..models import UnusualItem
 
+# 异动类型 → 粗粒度名称映射（Issue #62）。
+# 0x15/0x16/0x1D/0x1E 语义由 2026-09-01 全市场 12871 条实测锚定，详见
+# docs/protocol-unknown-fields.md「市场异动（0x1237）异动类型」一节。
+UNUSUAL_TYPE_NAMES: dict[int, str] = {
+    0x03: "主力买入卖出",
+    0x04: "加速拉升",
+    0x05: "加速下跌",
+    0x06: "低位反弹",
+    0x07: "高位回落",
+    0x08: "撑杆跳高",
+    0x09: "平台跳水",
+    0x0A: "单笔冲涨跌",
+    0x0B: "区间放量",
+    0x0C: "区间缩量",
+    0x10: "大单托盘",
+    0x11: "大单压盘",
+    0x12: "大单锁盘",
+    0x13: "竞价试盘",
+    0x14: "涨跌停",
+    0x15: "竞价/尾盘异动",
+    0x16: "盘中强势弱势",
+    0x1D: "急速拉升",
+    0x1E: "急速下跌",
+}
 
-def _describe_unusual(unusual_type: int, data: bytes) -> tuple[str, str]:
-    """根据异动类型解析描述和数值。"""
+
+def _describe_unusual(unusual_type: int, data: bytes, hour: int = 9) -> tuple[str, str]:
+    """根据异动类型解析描述和数值。hour 用于区分竞价/尾盘双时刻信号（0x15）。"""
     if len(data) < 13:
         return "", ""
     v1, v2, v3, v4 = struct.unpack_from("<B2fI", data)
@@ -56,8 +81,14 @@ def _describe_unusual(unusual_type: int, data: bytes) -> tuple[str, str]:
         desc = "大单锁盘"
         val = ""
     elif unusual_type == 0x13:
-        desc = "竞价试买"
-        val = f"{v2:.2f}/{v3:.2f}"
+        # 竞价试盘（09:15~09:20 触发）：v1=0x00 试买（申报价高于昨收）/ 0x01 试卖
+        # （低于昨收）；v2 为申报价，v3 为竞价量（手）。方向规律 2026-09-02
+        # 全量 552 条对照昨收 549 条一致（2 条恰等于昨收的边界 + 1 条异常）。
+        if v1 == 0x01:
+            desc = "竞价试卖"
+        else:
+            desc = "竞价试买"
+        val = f"{v2:.2f}/{v3:.0f}手"
     elif unusual_type == 0x14:
         direction = "涨" if v1 == 0x00 else "跌"
         if len(data) >= 10:
@@ -75,6 +106,31 @@ def _describe_unusual(unusual_type: int, data: bytes) -> tuple[str, str]:
         else:
             desc = f"涨跌停({direction})"
         val = f"{v2_alt:.2f}/{v3_alt:.2f}"
+    elif unusual_type == 0x15:
+        # 竞价/尾盘异动：开盘竞价（09:25）与收盘（15:00）两个撮合时刻都会触发。
+        # v1=0x02 拉升 / 0x03 下跌 / 0x01 平稳（±0.5% 分档）；v2 为时段尾段价格
+        # 变动（相对昨收），v3 为该时段成交量（手）。
+        stage = "竞价" if hour < 12 else "尾盘"
+        if v1 == 0x02:
+            desc = f"{stage}拉升"
+        elif v1 == 0x03:
+            desc = f"{stage}下跌"
+        elif v1 == 0x01:
+            desc = f"{stage}平稳"
+        else:
+            desc = f"{stage}异动"
+        val = f"{v2 * 100:.2f}%/{v3:.0f}手"
+    elif unusual_type == 0x16:
+        # 盘中强势/弱势：v2 = 触发时涨跌幅（09:25 样本与开盘涨幅 49/49 精确一致），
+        # v1 为带符号强弱等级（0x01~0x03 强势 1~3 级，0xFD~0xFF 弱势 1~3 级）。
+        desc = "盘中强势" if v2 >= 0 else "盘中弱势"
+        val = f"{v2 * 100:.2f}%"
+    elif unusual_type == 0x1D:
+        desc = "急速拉升"
+        val = f"{v2 * 100:.2f}%"
+    elif unusual_type == 0x1E:
+        desc = "急速下跌"
+        val = f"{v2 * 100:.2f}%"
     else:
         desc = f"异动类型{unusual_type:#04x}"
         val = ""
@@ -129,10 +185,10 @@ class UnusualCmd(BaseCommand[list[UnusualItem]]):
                 "<H6sBBBHH", body, offset, f"unusual record[{i}]"
             )
 
-            desc, value = _describe_unusual(unusual_type, body[offset + 15 : offset + 28])
-
             hour, minute_sec = unpack_from("<BH", body, offset + 29, f"unusual time[{i}]")
 
+            desc, value = _describe_unusual(unusual_type, body[offset + 15 : offset + 28], hour)
+
             results.append(
                 UnusualItem(
                     index=index,