02-组合风控与资金管理

预计学习时间：2.5 小时

难度：⭐⭐⭐

核心问题：给定风险度量结果，如何科学地分配资金和控制仓位？

资金管理全景

┌────────────────────────────────────────────────────────────────┐
│                    资金管理体系架构                              │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│                       资金管理                                  │
│                          │                                     │
│       ┌──────────────────┼──────────────────┐                 │
│       │                  │                  │                  │
│   仓位决策          风险预算           止损机制               │
│       │                  │                  │                  │
│  · Kelly 准则       · Risk Parity       · 硬止损              │
│  · 波动率调仓       · 等风险贡献         · 追踪止损            │
│  · ATR 仓位         · 多策略分配         · 回撤止损            │
│  · 动态杠杆         · 波动率目标         · 时间止损            │
│                                                                │
└────────────────────────────────────────────────────────────────┘

一、Kelly 准则

1.1 基本公式

Kelly 准则在 1956 年由贝尔实验室的 John Kelly 提出，用于在赌博和投资中最大化长期复利增长率。

对于仅有胜/负两种结果的简单情形：

$f^{*} = \frac{b p - q}{b} = p - \frac{q}{b}$

其中：

$f^{*}$ ：最优仓位比例
$b$ ：盈亏比（平均盈利 / 平均亏损）
$p$ ：胜率
$q = 1 - p$ ：败率

def kelly_fraction(win_rate, avg_win, avg_loss):
    """
    计算 Kelly 最优仓位比例
 
    参数
    ----
    win_rate : float
      胜率，如 0.55
    avg_win : float
      平均盈利（正数），如 0.02
    avg_loss : float
      平均亏损（正数），如 0.015
 
    返回
    ----
    kelly : float
      Kelly 最优仓位比例
    """
    b = avg_win / avg_loss  # 盈亏比
    q = 1 - win_rate
    kelly = (b * win_rate - q) / b
    return max(0, kelly)
 
# 示例
f_kelly = kelly_fraction(win_rate=0.55, avg_win=0.02, avg_loss=0.015)
print(f"Kelly 最优仓位: {f_kelly:.2%}")

1.2 分数 Kelly（Fractional Kelly）

实践中几乎不使用满 Kelly。半 Kelly 或四分之一 Kelly 是常见的折中方案。

Kelly 比例	长期增长率	风险（波动率）	破产风险
1.0x（满 Kelly）	最高	高	理论上为 0，但现实中极大
0.5x（半 Kelly）	75%	中	极低
0.25x（四分之一 Kelly）	44%	低	接近 0

def kelly_simulate(win_rate=0.55, avg_win=0.02, avg_loss=0.015,
                   kelly_frac=0.5, n_trades=1000, n_sim=1000):
    """
    模拟 Kelly 策略的长期表现
 
    参数
    ----
    win_rate : float
      胜率
    avg_win : float
      平均盈利
    avg_loss : float
      平均亏损
    kelly_frac : float
      Kelly 分数（如 0.5 表示半 Kelly）
    n_trades : int
      模拟交易次数
    n_sim : int
      蒙特卡洛模拟次数
 
    返回
    ----
    results : dict
      终值统计信息
    """
    f_star = kelly_fraction(win_rate, avg_win, avg_loss)
    f = f_star * kelly_frac
 
    # 蒙特卡洛模拟
    final_values = []
    equity_curves = []
 
    for _ in range(n_sim):
        equity = 1.0
        curve = [equity]
        for _ in range(n_trades):
            if np.random.random() < win_rate:
                # 盈利：仓位 f，盈利 avg_win
                equity *= (1 + f * avg_win / avg_win)  # 简化为 f
                equity *= (1 + avg_win * f)
            else:
                # 亏损：仓位 f，亏损 avg_loss
                equity *= (1 - avg_loss * f)
            curve.append(equity)
        final_values.append(equity)
        equity_curves.append(curve)
 
    final_values = np.array(final_values)
    return {
        'kelly_optimal': f_star,
        'actual_fraction': f,
        'median_final': np.median(final_values),
        'mean_final': np.mean(final_values),
        'p5_final': np.percentile(final_values, 5),
        'p95_final': np.percentile(final_values, 95),
        'ruin_rate': np.mean(final_values < 0.01),
    }
 
# 比较不同 Kelly 分数
print("=== Kelly 分数对比 ===")
for frac in [1.0, 0.5, 0.25]:
    res = kelly_simulate(kelly_frac=frac)
    print(f"  {frac:.0%} Kelly -> 中位终值: {res['median_final']:.2f}, "
          f"5%分位: {res['p5_final']:.2f}, "
          f"破产率: {res['ruin_rate']:.2%}")

1.3 连续收益率版 Kelly

对于连续收益率，Kelly 公式等价于：

$f^{*} = \frac{μ}{σ ^{2}}$

def kelly_continuous(returns):
    """
    连续收益率版 Kelly 公式
 
    参数
    ----
    returns : Series
      收益率序列
 
    返回
    ----
    kelly : float
      最优仓位比例
    """
    mu = returns.mean()      # 期望收益
    sigma2 = returns.var()   # 方差
    if sigma2 == 0:
        return 0.0
    return mu / sigma2
 
# 示例
np.random.seed(42)
returns = pd.Series(np.random.normal(0.001, 0.02, 252))
kelly = kelly_continuous(returns)
print(f"连续 Kelly 仓位: {kelly:.2%}")
print(f"半 Kelly 仓位: {kelly * 0.5:.2%}")

二、波动率调整仓位

2.1 核心思想

高波动时减少仓位，低波动时增加仓位，使每单位资金的波动贡献恒定。

$目标仓位 = \frac{目标波动率}{实际波动率} \times 基础仓位$

def volatility_target_position(returns, target_vol=0.10, lookback=60):
    """
    基于波动率目标的仓位调整
 
    参数
    ----
    returns : Series
      日收益率序列
    target_vol : float
      年化目标波动率
    lookback : int
      滚动窗口（交易日）
 
    返回
    ----
    positions : Series
      每日仓位比例序列
    """
    # 滚动计算年化波动率
    rolling_vol = returns.rolling(lookback).std() * np.sqrt(252)
 
    # 仓位 = 目标波动率 / 实际波动率
    positions = target_vol / rolling_vol
 
    # 限制仓位范围，避免极端值
    positions = positions.clip(0.1, 3.0)
 
    return positions
 
# 示例
np.random.seed(42)
returns = pd.Series(np.random.normal(0.0005, 0.02, 500))
 
positions = volatility_target_position(returns, target_vol=0.12)
print(f"仓位统计:")
print(f"  均值: {positions.mean():.2f}")
print(f"  中位数: {positions.median():.2f}")
print(f"  最小值: {positions.min():.2f}")
print(f"  最大值: {positions.max():.2f}")

2.2 ATR 仓位管理

ATR（Average True Range）衡量价格的真实波动幅度，常用于趋势跟踪策略的仓位计算。

$股数 = \frac{风险金额}{A T R \times 合约乘数}$

def atr_position_size(high, low, close, risk_per_trade=0.02,
                      risk_amount=None, atr_period=14):
    """
    基于 ATR 的仓位计算
 
    参数
    ----
    high, low, close : Series
      最高价、最低价、收盘价序列
    risk_per_trade : float
      单笔交易风险比例（占总资金）
    risk_amount : float
      单笔风险金额（若指定则覆盖 risk_per_trade）
    atr_period : int
      ATR 计算周期
 
    返回
    ----
    position_units : Series
      建议持仓单位数
    atr : Series
      ATR 序列
    """
    # 计算 True Range
    tr1 = high - low
    tr2 = abs(high - close.shift(1))
    tr3 = abs(low - close.shift(1))
    tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
 
    # 计算 ATR
    atr = tr.rolling(atr_period).mean()
 
    # 计算仓位
    if risk_amount is None:
        # 使用最近收盘价估算总资金
        total_capital = 1_000_000  # 假设 100 万资金
        risk_amount = total_capital * risk_per_trade
 
    # 仓位 = 风险金额 / ATR
    position_units = risk_amount / atr
 
    return position_units, atr
 
# 示例
np.random.seed(42)
n = 100
close = pd.Series(100 + np.cumsum(np.random.normal(0, 0.5, n)))
high = close * (1 + np.abs(np.random.normal(0.01, 0.005, n)))
low = close * (1 - np.abs(np.random.normal(0.01, 0.005, n)))
 
units, atr = atr_position_size(high, low, close)
print(f"最新 ATR: {atr.iloc[-1]:.2f}")
print(f"建议持仓单位: {units.iloc[-1]:.0f}")

三、风险预算

3.1 Risk Parity（风险平价）

Risk Parity 让每个资产对组合总风险的贡献相等，而非让每个资产的资金权重相等。

每个资产的风险贡献（RC）：

$R C_{i} = w_{i} \frac{( Σ w ) _{i}}{w ^{⊤} Σ w}$

目标： $R C_{1} = R C_{2} = \dots = R C_{N}$

def risk_parity(returns, target_risk_contrib=None):
    """
    Risk Parity 权重优化
 
    参数
    ----
    returns : DataFrame
      资产收益率矩阵 (T x N)
    target_risk_contrib : list
      目标风险贡献比例，如 [0.5, 0.3, 0.2]
      若为 None，则等风险贡献
 
    返回
    ----
    weights : ndarray
      优化后的权重
    risk_contribs : Series
      各资产风险贡献比例
    """
    cov = np.cov(returns, rowvar=False, ddof=1)
    n = cov.shape[0]
 
    if target_risk_contrib is None:
        target_risk_contrib = np.ones(n) / n
 
    # 目标函数：最小化 (RC_i - RC_j)^2
    def objective(w):
        sigma_p = np.sqrt(w @ cov @ w)
        marginal = cov @ w / sigma_p  # 边际风险贡献
        rc = w * marginal               # 风险贡献
        # 目标：RC 比例 = target_risk_contrib
        rc_ratio = rc / rc.sum()
        return np.sum((rc_ratio - target_risk_contrib) ** 2)
 
    # 约束：权重和为 1，权重为正
    constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
    bounds = [(1e-6, 1.0)] * n
    w0 = np.ones(n) / n
 
    result = minimize(objective, w0, method='SLSQP',
                      bounds=bounds, constraints=constraints,
                      options={'maxiter': 1000, 'ftol': 1e-15})
 
    weights = result.x
 
    # 计算实际风险贡献
    sigma_p = np.sqrt(weights @ cov @ weights)
    marginal = cov @ weights / sigma_p
    rc = weights * marginal
    rc_ratio = rc / rc.sum()
 
    return weights, pd.Series(rc_ratio, index=returns.columns)
 
# 示例
np.random.seed(42)
n_days, n_assets = 500, 4
asset_names = ['国内股票', '海外股票', '国内债券', '黄金']
asset_returns = pd.DataFrame(
    np.column_stack([
        np.random.normal(0.0008, 0.020, n_days),  # 股票高波动
        np.random.normal(0.0005, 0.018, n_days),  # 海外股票
        np.random.normal(0.0003, 0.005, n_days),  # 债券低波动
        np.random.normal(0.0002, 0.012, n_days),  # 黄金
    ]),
    columns=asset_names
)
 
rp_weights, rp_rc = risk_parity(asset_returns)
print("Risk Parity 权重:")
for name, w in zip(asset_names, rp_weights):
    print(f"  {name}: {w:.2%}")
print(f"\n风险贡献比例:")
print(rp_rc.round(4))

3.2 等风险贡献（ERC）

ERC 是 Risk Parity 的特例，要求每个资产的风险贡献完全相等。上面的代码中 target_risk_contrib=None 即为 ERC。

3.3 自定义风险预算

# 示例：自定义风险预算 — 股票承担 50% 风险，债券 30%，黄金 20%
custom_weights, custom_rc = risk_parity(
    asset_returns, target_risk_contrib=[0.5, 0.2, 0.2, 0.1]
)
print("\n自定义风险预算权重:")
for name, w in zip(asset_names, custom_weights):
    print(f"  {name}: {w:.2%}")
print(f"风险贡献比例: {custom_rc.round(3).to_dict()}")

四、止损机制

4.1 三种止损策略

价格走势与止损示意

价格
  │    ╱╲    ╱╲
  │   ╱  ╲  ╱  ╲╱╲
  │  ╱    ╲╱      ╲
  │ ╱               ╲  ← 追踪止损线随价格上升
  │╱────────────────╲
  │                   ╲╱  ← 触发追踪止损
  │
  ├─── 硬止损（固定）
  ├─── 回撤止损（从最高点回撤 X%）
  └─── 追踪止损（随最高价浮动）

class StopLossManager:
    """
    多策略止损管理器
 
    支持：硬止损、追踪止损、回撤止损、时间止损
    """
 
    def __init__(self, method='trailing', **params):
        """
        参数
        ----
        method : str
          'hard' | 'trailing' | 'drawdown' | 'time'
        params : dict
          止损参数
        """
        self.method = method
        self.params = params
        self.max_price = None  # 追踪止损用
        self.peak_equity = None  # 回撤止损用
        self.entry_time = None  # 时间止损用
        self.stopped = False
 
    def check(self, current_price, current_equity=None, current_time=None):
        """
        检查是否触发止损
 
        参数
        ----
        current_price : float
          当前价格
        current_equity : float
          当前权益（回撤止损用）
        current_time : int
          当前时间步（时间止损用）
 
        返回
        ----
        should_stop : bool
          是否触发止损
        reason : str
          触发原因
        """
        if self.method == 'hard':
            return self._check_hard(current_price)
 
        elif self.method == 'trailing':
            return self._check_trailing(current_price)
 
        elif self.method == 'drawdown':
            return self._check_drawdown(current_equity)
 
        elif self.method == 'time':
            return self._check_time(current_time)
 
        return False, "unknown"
 
    def _check_hard(self, price):
        """硬止损：价格低于固定阈值"""
        stop_price = self.params.get('stop_price')
        if stop_price is None:
            stop_pct = self.params.get('stop_pct', 0.05)
            entry = self.params['entry_price']
            stop_price = entry * (1 - stop_pct)
        should_stop = price <= stop_price
        return should_stop, f"硬止损触发 (价格 {price:.2f} <= {stop_price:.2f})"
 
    def _check_trailing(self, price):
        """追踪止损：从最高价回撤一定比例"""
        if self.max_price is None:
            self.max_price = price
 
        self.max_price = max(self.max_price, price)
        trail_pct = self.params.get('trail_pct', 0.08)
        stop_price = self.max_price * (1 - trail_pct)
 
        should_stop = price <= stop_price
        return should_stop, f"追踪止损触发 (从最高 {self.max_price:.2f} 回撤)"
 
    def _check_drawdown(self, equity):
        """回撤止损：权益从高点回撤一定比例"""
        if self.peak_equity is None:
            self.peak_equity = equity
 
        self.peak_equity = max(self.peak_equity, equity)
        max_dd_pct = self.params.get('max_dd_pct', 0.10)
        current_dd = 1 - equity / self.peak_equity
 
        should_stop = current_dd >= max_dd_pct
        return should_stop, f"回撤止损触发 (回撤 {current_dd:.2%} >= {max_dd_pct:.2%})"
 
    def _check_time(self, time):
        """时间止损：持仓超过指定时间"""
        if self.entry_time is None:
            self.entry_time = time
 
        max_hold = self.params.get('max_hold', 60)
        should_stop = (time - self.entry_time) >= max_hold
        return should_stop, f"时间止损触发 (持仓 {time - self.entry_time} >= {max_hold})"
 
# 示例：模拟价格序列，测试追踪止损
np.random.seed(42)
prices = pd.Series(
    [100, 102, 105, 103, 108, 112, 110, 115, 118, 113, 108, 104]
)
 
sl = StopLossManager(method='trailing', trail_pct=0.07)
print("=== 追踪止损测试 ===")
for i, price in enumerate(prices):
    should_stop, reason = sl.check(price)
    status = "触发!" if should_stop else "持仓"
    print(f"  第{i}天: 价格={price}, 最高={sl.max_price}, {status}")
    if should_stop:
        print(f"    -> {reason}")
        break

4.2 止损策略对比

止损类型	优势	劣势	适用场景
硬止损	简单明确	容易被震荡洗出	短线交易
追踪止损	保护利润	回撤比例难设定	趋势跟踪
回撤止损	控制最大损失	反应较慢	组合层面
时间止损	避免资金闲置	可能过早退出	均值回归策略

五、动态杠杆调整

5.1 原理

根据市场环境和策略表现动态调整杠杆，在有利时放大，不利时收缩。

$L_{t} = L_{0} \times \frac{σ _{t a r g e t}}{σ _{t, r o l l in g}} \times regime_adjustment$

def dynamic_leverage(returns, base_leverage=1.5, target_vol=0.15,
                     lookback=60, max_leverage=3.0, min_leverage=0.3):
    """
    动态杠杆调整
 
    参数
    ----
    returns : Series
      策略日收益率
    base_leverage : float
      基础杠杆
    target_vol : float
      目标年化波动率
    lookback : int
      滚动窗口
    max_leverage : float
      最大杠杆
    min_leverage : float
      最小杠杆
 
    返回
    ----
    leverage : Series
      每日杠杆序列
    leveraged_returns : Series
      加杠杆后的收益率
    """
    # 滚动波动率
    rolling_vol = returns.rolling(lookback).std() * np.sqrt(252)
 
    # 根据波动率调整杠杆
    leverage = base_leverage * (target_vol / rolling_vol)
 
    # 限制杠杆范围
    leverage = leverage.clip(min_leverage, max_leverage)
 
    # 加杠杆后的收益
    leveraged_returns = returns * leverage.shift(1)
 
    return leverage, leveraged_returns
 
# 示例
np.random.seed(42)
returns = pd.Series(np.random.normal(0.001, 0.015, 500))
leverage, lev_returns = dynamic_leverage(returns)
 
print(f"杠杆统计:")
print(f"  均值: {leverage.mean():.2f}")
print(f"  最大值: {leverage.max():.2f}")
print(f"  最小值: {leverage.min():.2f}")
print(f"\n无杠杆年化收益: {returns.mean() * 252:.2%}")
print(f"有杠杆年化收益: {lev_returns.mean() * 252:.2%}")
print(f"无杠杆年化波动: {returns.std() * np.sqrt(252):.2%}")
print(f"有杠杆年化波动: {lev_returns.std() * np.sqrt(252):.2%}")

六、波动率目标策略

波动率目标是机构最常用的风控框架之一。核心思想：维持组合波动率在目标水平附近。

def volatility_targeting(returns, target_vol=0.12, lookback=60,
                          max_vol=0.30, rebalance_freq=5):
    """
    波动率目标策略
 
    参数
    ----
    returns : Series
      策略收益率
    target_vol : float
      目标年化波动率
    lookback : int
      波动率估计窗口
    max_vol : float
      波动率上限（若超过则减仓至零）
    rebalance_freq : int
      再平衡频率（天）
 
    返回
    ----
    target_returns : Series
      波动率目标调整后的收益
    vol_scale : Series
      波动率缩放因子
    realized_vol : Series
      实现波动率
    """
    # 滚动年化波动率
    realized_vol = returns.rolling(lookback).std() * np.sqrt(252)
 
    # 计算缩放因子
    vol_scale = target_vol / realized_vol
 
    # 波动率过高时减仓
    vol_scale = vol_scale.where(realized_vol <= max_vol, 0)
 
    # 限制缩放范围
    vol_scale = vol_scale.clip(0, 3.0)
 
    # 按频率再平衡
    vol_scale = vol_scale.resample(f'{rebalance_freq}D').last().ffill()
 
    # 调整后收益
    target_returns = returns * vol_scale.shift(1)
 
    return target_returns, vol_scale, realized_vol
 
# 示例
np.random.seed(42)
returns = pd.Series(np.random.normal(0.001, 0.02, 500),
                     index=pd.bdate_range('2023-01-01', periods=500))
 
tgt_ret, scale, real_vol = volatility_targeting(returns, target_vol=0.10)
print(f"调整前年化波动率: {returns.std() * np.sqrt(252):.2%}")
print(f"调整后年化波动率: {tgt_ret.std() * np.sqrt(252):.2%}")
print(f"平均缩放因子: {scale.mean():.2f}")

七、多策略资金分配

7.1 核心问题

当同时运行多个策略时，如何在它们之间分配资金？

def multi_strategy_allocation(strategy_returns, method='equal_risk',
                               total_capital=1e7):
    """
    多策略资金分配
 
    参数
    ----
    strategy_returns : DataFrame
      各策略日收益率 (T x K)
    method : str
      分配方法: 'equal_weight' | 'equal_risk' | 'sharpe_weight'
    total_capital : float
      总资金
 
    返回
    ----
    allocations : Series
      各策略资金分配
    """
    n_strategies = strategy_returns.shape[1]
 
    if method == 'equal_weight':
        # 等权分配
        weights = np.ones(n_strategies) / n_strategies
 
    elif method == 'equal_risk':
        # 等风险分配：按波动率倒数加权
        vols = strategy_returns.std()
        inv_vols = 1 / vols
        weights = inv_vols / inv_vols.sum()
 
    elif method == 'sharpe_weight':
        # 按 Sharpe 比率加权（仅正 Sharpe 策略）
        sharpe = strategy_returns.mean() / strategy_returns.std() * np.sqrt(252)
        sharpe = sharpe.clip(lower=0)  # 负 Sharpe 策略权重为 0
        weights = sharpe / sharpe.sum()
 
    else:
        raise ValueError(f"未知的分配方法: {method}")
 
    allocations = pd.Series(weights * total_capital, index=strategy_returns.columns)
    return allocations
 
# 示例
np.random.seed(42)
n_days = 500
strategies = {
    '动量策略': np.random.normal(0.0008, 0.015, n_days),
    '均值回归': np.random.normal(0.0006, 0.010, n_days),
    '统计套利': np.random.normal(0.0004, 0.008, n_days),
    '趋势跟踪': np.random.normal(0.0005, 0.018, n_days),
}
strategy_returns = pd.DataFrame(strategies)
 
print("=== 多策略资金分配 ===")
for method in ['equal_weight', 'equal_risk', 'sharpe_weight']:
    alloc = multi_strategy_allocation(strategy_returns, method)
    print(f"\n{method}:")
    for name, val in alloc.items():
        print(f"  {name}: {val:,.0f} 元 ({val / 1e7:.1%})")

7.2 策略相关性管理

def strategy_correlation_risk(strategy_returns, max_corr=0.5):
    """
    识别高相关性策略对
 
    参数
    ----
    strategy_returns : DataFrame
      策略收益率矩阵
    max_corr : float
      相关性阈值
 
    返回
    ----
    high_corr_pairs : list
      高相关性策略对
    """
    corr_matrix = strategy_returns.corr()
    n = len(strategy_returns.columns)
    high_corr_pairs = []
 
    for i in range(n):
        for j in range(i + 1, n):
            corr_val = corr_matrix.iloc[i, j]
            if abs(corr_val) >= max_corr:
                high_corr_pairs.append({
                    '策略1': strategy_returns.columns[i],
                    '策略2': strategy_returns.columns[j],
                    '相关性': round(corr_val, 4)
                })
 
    return high_corr_pairs, corr_matrix
 
# 示例
pairs, corr = strategy_correlation_risk(strategy_returns)
print("\n策略相关性矩阵:")
print(corr.round(3))
if pairs:
    print(f"\n高相关性策略对 (|corr| >= 0.5):")
    for p in pairs:
        print(f"  {p['策略1']} - {p['策略2']}: {p['相关性']}")
else:
    print("\n未发现高相关性策略对")

八、PositionManager 综合类

下面将仓位管理、止损、杠杆等功能整合为统一接口。

class PositionManager:
    """
    仓位管理器
 
    功能：Kelly 仓位、波动率调整、风险预算、止损管理、动态杠杆
    """
 
    def __init__(self, initial_capital=1e7, max_leverage=2.0,
                 kelly_fraction=0.5, target_vol=0.15):
        """
        初始化
 
        参数
        ----
        initial_capital : float
          初始资金
        max_leverage : float
          最大杠杆
        kelly_fraction : float
          Kelly 分数
        target_vol : float
          目标年化波动率
        """
        self.initial_capital = initial_capital
        self.max_leverage = max_leverage
        self.kelly_fraction = kelly_fraction
        self.target_vol = target_vol
 
        # 状态
        self.capital = initial_capital
        self.positions = {}  # 资产 -> 持仓信息
        self.equity_curve = []
        self.trade_log = []
 
    def kelly_position(self, win_rate, avg_win, avg_loss,
                       max_position_pct=0.25):
        """
        Kelly 仓位计算
 
        参数
        ----
        win_rate : float
          胜率
        avg_win : float
          平均盈利
        avg_loss : float
          平均亏损
        max_position_pct : float
          单资产最大仓位占比
 
        返回
        ----
        position_pct : float
          建议仓位占总资金比例
        """
        kelly = kelly_fraction(win_rate, avg_win, avg_loss)
        kelly *= self.kelly_fraction  # 应用分数 Kelly
 
        # 限制最大仓位
        position_pct = min(kelly, max_position_pct)
        return position_pct
 
    def vol_adjusted_position(self, returns, asset_vol, capital=None):
        """
        波动率调整仓位
 
        参数
        ----
        returns : Series
          策略历史收益
        asset_vol : float
          资产当前年化波动率
        capital : float
          可用资金
 
        返回
        ----
        position_size : float
          建议仓位金额
        """
        if capital is None:
            capital = self.capital
 
        if asset_vol <= 0:
            return 0
 
        # 目标仓位 = 总资金 * 目标波动率 / 资产波动率
        raw_position = capital * (self.target_vol / asset_vol)
 
        # 应用杠杆限制
        max_position = capital * self.max_leverage
        position_size = min(raw_position, max_position)
 
        return position_size
 
    def risk_budget_position(self, returns, budgets):
        """
        风险预算仓位
 
        参数
        ----
        returns : DataFrame
          资产收益率
        budgets : dict
          资产 -> 目标风险预算
 
        返回
        ----
        positions : dict
          资产 -> 仓位金额
        """
        target_rc = [budgets[c] for c in returns.columns]
        weights, _ = risk_parity(returns, target_rc)
 
        positions = {}
        for col, w in zip(returns.columns, weights):
            positions[col] = w * self.capital
 
        return positions
 
    def apply_stop_loss(self, positions, prices, method='trailing',
                        **params):
        """
        对所有持仓应用止损检查
 
        参数
        ----
        positions : dict
          资产 -> 持仓信息（含 entry_price 等）
        prices : dict
          资产 -> 当前价格
        method : str
          止损方法
        params : dict
          止损参数
 
        返回
        ----
        actions : dict
          资产 -> 操作（'hold' 或 'stop'）
        """
        actions = {}
        for asset, info in positions.items():
            price = prices.get(asset)
            if price is None:
                actions[asset] = 'hold'
                continue
 
            sl = StopLossManager(method=method, **params, **info)
            should_stop, reason = sl.check(price)
            actions[asset] = 'stop' if should_stop else 'hold'
 
            if should_stop:
                self.trade_log.append({
                    'asset': asset,
                    'action': 'stop_loss',
                    'reason': reason,
                    'price': price
                })
 
        return actions
 
    def portfolio_summary(self):
        """生成组合摘要"""
        return {
            '初始资金': self.initial_capital,
            '当前资金': self.capital,
            '收益率': (self.capital / self.initial_capital - 1),
            '持仓数量': len(self.positions),
            '最大杠杆': self.max_leverage,
            'Kelly 分数': self.kelly_fraction,
            '目标波动率': self.target_vol,
        }
 
# 使用示例
pm = PositionManager(initial_capital=1e7, kelly_fraction=0.5,
                      target_vol=0.12)
 
# Kelly 仓位
pos_kelly = pm.kelly_position(win_rate=0.55, avg_win=0.02, avg_loss=0.015)
print(f"Kelly 仓位比例: {pos_kelly:.2%}")
 
# 波动率调整仓位
pos_vol = pm.vol_adjusted_position(returns, asset_vol=0.20)
print(f"波动率调整仓位: {pos_vol:,.0f} 元")
 
# 组合摘要
print(f"\n组合摘要: {pm.portfolio_summary()}")

总结

管理维度	核心工具	关键要点
仓位大小	Kelly / 波动率调整	永远用分数 Kelly
风险分配	Risk Parity / ERC	风险贡献比资金权重更重要
止损保护	追踪止损 / 回撤止损	组合层面用回撤止损
杠杆控制	动态杠杆 / 波动率目标	高波动时降杠杆
多策略	等风险分配	关注策略间相关性

核心原则：仓位管理的首要目标不是最大化收益，而是确保你在最坏情况下仍能留在牌桌上。活下来，才有机会赢。

下一步：学习03-绩效归因，学会拆解收益来源，持续优化策略。

MindCarver Blog

MindCarver

探索

02-组合风控与资金管理

02-组合风控与资金管理

资金管理全景

一、Kelly 准则

1.1 基本公式

1.2 分数 Kelly（Fractional Kelly）

1.3 连续收益率版 Kelly

二、波动率调整仓位

2.1 核心思想

2.2 ATR 仓位管理

三、风险预算

3.1 Risk Parity（风险平价）

3.2 等风险贡献（ERC）

3.3 自定义风险预算

四、止损机制

4.1 三种止损策略

4.2 止损策略对比

五、动态杠杆调整

5.1 原理

六、波动率目标策略

七、多策略资金分配

7.1 核心问题

7.2 策略相关性管理

八、PositionManager 综合类

总结

关系图谱

目录

反向链接