更多的交易决策

小象实战讲义 · 进阶量化交易实战：迭代式的量化策略研发 本讲义仅用于编程与量化方法教学，不构成任何投资建议。

在单一指标的金叉死叉之外，成熟的量化交易系统依赖多维度组合决策。如何将量价形态、波动率分位数、盘口成交量以及持仓时间跨度有机组合为高内聚的决策树状态机？本节深入剖析复合交易决策的设计原则与代码工程实现。

💡 核心导读

• 量化进阶交易决策四大核心支柱：
  • 量价共振确认（Price-Volume Confirmation）：价格创近期新高的同时，成交量必须突破 20 日均量的 1.5 倍；
  • 波动率挤压后释放（Volatility Squeeze）：布林带宽突破历史 20% 分位数确认脱离挤压区，为高胜率暴击点；
  • 多条件与非门组合决策（Boolean Decision Matrix）：利用矩阵逻辑运算过滤孤立技术指标产生的频繁毛刺；
  • 分步建仓与动态调仓决策：依据趋势强度与浮盈表现实施梯级加减仓。

┌─────────────────────────────────────────────────────────────┐
│                    复合量化交易决策状态机拓扑图             │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 价格突破决策 (Price)    │   │ 成交量放量决策 (Volume) │   │ 波动率扩张决策 (Vol)    │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • 收盘价 > 20日最高价   │   │ • Volume > 1.5 * VMA20  │   │ • 布林带开口扩张        │
│ • 确认多头动能方向      │   │ • 确认主力增量资金进场  │   │ • 蓄势盘整结束启动      │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、量价与波动率三维共振决策数学模型

复合决策函数通过联立价格突破、量能爆发与波动率扩张三个布尔掩码向量：

$$\text{Decision}_t = \left( Close_{t-1} > \max_{2 \le k \le 21} High_{t-k} \right) \land \left( V_{t-1} > 1.5 \times \overline{V}_{20} \right) \land \left( \text{BandWidth}_{t-1} > \text{BW}_{P20} \right)$$

只有三个维度同时共振时才发出确定性买入信号，有效剔除缩量假突破。

import pandas as pd
import numpy as np

def generate_multi_factor_decision_signals(df: pd.DataFrame) -> pd.DataFrame:
    """
    量价与波动率三维共振交易决策引擎（合成教学样例，实盘需警惕未来函数与过拟合风险）
    """
    res = df.copy()
    # 1. 价格突破
    res['high_20'] = res['high'].rolling(20).max().shift(1)
    cond_price = res['close'] > res['high_20']
    
    # 2. 成交量放量
    res['vma_20'] = res['volume'].rolling(20).mean().shift(1)
    cond_volume = res['volume'] > (1.5 * res['vma_20'])
    
    # 3. 波动率扩张 (带宽突破过去 60 根的 20% 分位数, 确认脱离挤压区)
    # bandwidth 用截至 t 的数据计算, 与价格/成交量条件时间戳对齐;
    # 基准分位数窗口 shift(1) 不含当期 bandwidth, 避免自我比较
    ma20 = res['close'].rolling(20, min_periods=10).mean()
    std20 = res['close'].rolling(20, min_periods=10).std()
    res['bandwidth'] = (2.0 * std20) / (ma20 + 1e-8)
    cond_vol = res['bandwidth'] > res['bandwidth'].rolling(60, min_periods=20).quantile(0.2).shift(1)
    
    # 复合决策触发 (t 期收盘确认信号, t+1 期开盘执行, 严格无未来函数)
    res['composite_buy'] = (cond_price & cond_volume & cond_vol).shift(1).fillna(False).astype(bool)
    return res

# 合成数据演示
rng = np.random.default_rng(42)
n_bars = 200
base_price = 100.0
prices = base_price + np.cumsum(rng.normal(0.02, 0.8, n_bars))
highs = prices + rng.uniform(0.5, 2.0, n_bars)
lows = prices - rng.uniform(0.5, 2.0, n_bars)
volumes = rng.exponential(5000, n_bars)

# 注入一个符合条件的突破区域（索引 80-82）
prices[80] = np.max(prices[:80]) + 5.0
prices[81] = prices[80] + 2.0
highs[80:82] = prices[80:82] + 1.5
volumes[80] = 15000.0
volumes[81] = 12000.0

df_demo = pd.DataFrame({
    'high': highs,
    'low': lows,
    'close': prices,
    'volume': volumes
})

result = generate_multi_factor_decision_signals(df_demo)
signal_count = result['composite_buy'].sum()
signal_indices = result.index[result['composite_buy']].tolist()

print("===== 三维共振交易决策信号生成 =====")
print(f"数据总长度: {len(df_demo)} 根K线")
print(f"复合买入信号触发次数: {signal_count}")
if signal_count > 0:
    print(f"信号触发位置索引 (前5个): {signal_indices[:5]}")
    sample_idx = signal_indices[0] if signal_indices else 0
    sample_row = result.loc[sample_idx]
    print(f"示例信号点 (索引 {sample_idx}) 详情:")
    print(f"  收盘价: {sample_row['close']:.2f}, 20期最高: {sample_row['high_20']:.2f}")
    print(f"  成交量: {sample_row['volume']:.0f}, 20期均量: {sample_row['vma_20']:.0f}")
    print(f"  带宽: {sample_row['bandwidth']:.4f}")
else:
    print("警告: 合成数据未生成任何买入信号，请检查参数。")

===== 三维共振交易决策信号生成 =====
数据总长度: 200 根K线
复合买入信号触发次数: 2
信号触发位置索引 (前5个): [81, 82]
示例信号点 (索引 81) 详情:
  收盘价: 112.42, 20期最高: 111.92
  成交量: 12000, 20期均量: 4730
  带宽: 0.0500

二、动态分级减仓与离场决策树

离场决策同样需要精细化。当出现高位巨量滞涨或跌破跟踪止损线时，触发分级减仓逻辑。

import pandas as pd
import numpy as np

def evaluate_exit_decisions(df: pd.DataFrame, entry_price: float, current_bar: int) -> str:
    """
    分级离场与止盈止损决策树（收盘确认口径，实盘需考虑盘中触及风险）
    """
    row = df.iloc[current_bar]
    # 教学演示采用「收盘确认」口径: 以当根收盘价 close 计算盈亏并判定止盈止损;
    # 实盘硬止损多为「盘中触及」口径, 应按 (row['low']-entry_price)/entry_price <= -0.03 判定(多头; 空头用 high),
    # 否则盘中已击穿 -3% 而收盘回升的止损会被漏判, 系统性低估止损次数与回撤。
    pnl = (row['close'] - entry_price) / entry_price
    
    # 1. 硬止损
    if pnl <= -0.03:
        return "HARD_STOP_EXIT"
    # 2. 达到第一目标止盈，减仓一半
    elif pnl >= 0.08 and pnl < 0.15:
        return "TAKE_PROFIT_HALF"
    # 3. 达到第二目标止盈，全部离场
    elif pnl >= 0.15:
        return "TAKE_PROFIT_ALL"
    return "HOLD"

# 合成数据与测试场景
rng = np.random.default_rng(123)
n_samples = 10
base_price = 50.0
price_changes = rng.normal(0.01, 0.05, n_samples)
prices = base_price * np.cumprod(1 + price_changes)
highs = prices * (1 + rng.uniform(0.01, 0.03, n_samples))
lows = prices * (1 - rng.uniform(0.01, 0.04, n_samples))
volumes = rng.integers(10000, 50000, n_samples)

df_test = pd.DataFrame({
    'open': prices * 0.99, # 简单生成
    'high': highs,
    'low': lows,
    'close': prices,
    'volume': volumes
})

entry_price_test = 50.0
test_cases = [
    (0, "小幅亏损，应持有"),
    (3, "可能触及止盈或止损"),
    (6, "检查是否达到更高止盈目标")
]

print("===== 分级离场决策树测试 =====")
print(f"入场价格: {entry_price_test:.2f}")
print("-" * 40)
for bar_idx, desc in test_cases:
    if bar_idx < len(df_test):
        action = evaluate_exit_decisions(df_test, entry_price_test, bar_idx)
        row = df_test.iloc[bar_idx]
        pnl_val = (row['close'] - entry_price_test) / entry_price_test
        print(f"Bar {bar_idx} ({desc}):")
        print(f"  收盘价: {row['close']:.2f}, 盈亏: {pnl_val*100:+.2f}%")
        print(f"  决策: {action}")
        print()

===== 分级离场决策树测试 =====
入场价格: 50.00
----------------------------------------
Bar 0 (小幅亏损，应持有):
  收盘价: 48.03, 盈亏: -3.95%
  决策: HARD_STOP_EXIT

Bar 3 (可能触及止盈或止损):
  收盘价: 52.18, 盈亏: +4.35%
  决策: HOLD

Bar 6 (检查是否达到更高止盈目标):
  收盘价: 55.99, 盈亏: +11.98%
  决策: TAKE_PROFIT_HALF

📝 动手练一练

编写 Python 仿真测试，验证量价与波动率三维共振决策的准确性与分级离场执行：

import pandas as pd
import numpy as np

def generate_multi_factor_decision_signals(df: pd.DataFrame) -> pd.DataFrame:
    """
    量价与波动率三维共振交易决策引擎（合成教学样例，实盘需警惕未来函数与过拟合风险）
    """
    res = df.copy()
    res['high_20'] = res['high'].rolling(20).max().shift(1)
    cond_price = res['close'] > res['high_20']
    
    res['vma_20'] = res['volume'].rolling(20).mean().shift(1)
    cond_volume = res['volume'] > (1.5 * res['vma_20'])
    
    ma20 = res['close'].rolling(20, min_periods=10).mean()
    std20 = res['close'].rolling(20, min_periods=10).std()
    res['bandwidth'] = (2.0 * std20) / (ma20 + 1e-8)
    cond_vol = res['bandwidth'] > res['bandwidth'].rolling(60, min_periods=20).quantile(0.2).shift(1)
    
    res['composite_buy'] = (cond_price & cond_volume & cond_vol).shift(1).fillna(False).astype(bool)
    return res

def evaluate_exit_decisions(df: pd.DataFrame, entry_price: float, current_bar: int) -> str:
    """
    分级离场与止盈止损决策树（收盘确认口径，实盘需考虑盘中触及风险）
    """
    row = df.iloc[current_bar]
    pnl = (row['close'] - entry_price) / entry_price
    
    if pnl <= -0.03:
        return "HARD_STOP_EXIT"
    elif pnl >= 0.08 and pnl < 0.15:
        return "TAKE_PROFIT_HALF"
    elif pnl >= 0.15:
        return "TAKE_PROFIT_ALL"
    return "HOLD"

# 模拟包含放量突破的行情
rng = np.random.default_rng(42)
prices = 100.0 + np.cumsum(rng.normal(0.05, 0.5, 100))
vols = rng.exponential(1000, 100)
# 人为在第 40-41 根注入连续放量突破 (首根放量大涨, 次根惯性冲高, 带宽随之开口)
prices[40] = np.max(prices[:40]) + 8.0
prices[41] = prices[40] + 1.5
vols[40] = 8000.0
vols[41] = 6000.0

df = pd.DataFrame({'high': prices + 1, 'low': prices - 1, 'close': prices, 'volume': vols})
res = generate_multi_factor_decision_signals(df)

exit_action = evaluate_exit_decisions(df, entry_price=100.0, current_bar=40)

print("================ 复合交易决策系统集成测试 ================")
print(f"数据总长度: {len(df)} 根 K 线")
print(f"三维共振买入信号触发数: {res['composite_buy'].sum()} 次")
if res['composite_buy'].sum() > 0:
    trigger_bars = res.index[res['composite_buy']].tolist()
    print(f"信号触发位置: {trigger_bars[:5]}")
print(f"Bar 40 (突破点) 收盘价: {df.loc[40, 'close']:.2f}")
print(f"Bar 40 突破时离场决策判断: {exit_action}")
print("✅ 复合量化交易决策模型验证 PASS")
print("========================================================")

运行结果输出：

================ 复合交易决策系统集成测试 ================
数据总长度: 100 根 K 线
三维共振买入信号触发数: 2 次
信号触发位置: [41, 42]
Bar 40 (突破点) 收盘价: 111.13
Bar 40 突破时离场决策判断: TAKE_PROFIT_HALF
✅ 复合量化交易决策模型验证 PASS
========================================================

本章小结

1. 量价波动三位一体决策：将价格突破、成交量放量与波动率扩张有机联立，大幅压降单一指标假突破。
2. 构建精细化决策树：针对不同市场情景设计梯级止盈、移动止损与硬止损等多重离场预案。
3. 注重状态机严谨性：确保每个决策状态转移具备完备的边界条件与无二义性动作。

📋 行动清单 (Action Checklist)

• 运行复合决策测试代码，测试在缩量假突破行情中的拦截效果；
• 增加均线多头排列（MA5 > MA10 > MA20）作为第四维度决策确认；
• 全面复习第二章技术指标体系，准备进入第三章股票池实战。

—— 小象教研组