用python编程实现资金管理模块

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

将资金管理理论具象化为高内聚、易扩展的量化代码是工程落地的关键。本节深入剖析头寸管理基类抽象、风险平价分配器（Risk Parity Sizer）以及动态杠杆限制器的 Python 3.12 模块化实现。

💡 核心导读

• 资金管理模块架构设计的三大核心原则：
  • 抽象基类（Abstract Base Class）：定义标准的 calculate_order_size() 统一契约接口；
  • 多标的资金预算协调（Portfolio Budget Manager）：防止多个信号同时触发导致账户可用资金超限；
  • 实时购买力与保证金折算（Buying Power & Margin Check）：精准扣除期货保证金与现货可用资金。

┌─────────────────────────────────────────────────────────────┐
│                    资金管理模块架构与数据流拓扑             │
└──────────────────────────────┬──────────────────────────────┘
                               │
 ┌─────────────────────────────┼─────────────────────────────┐
 ▼                             ▼                             ▼
┌─────────────────────────┐   ┌─────────────────────────┐   ┌─────────────────────────┐
│ 策略信号生成            │   │ 资金管理协调器 (Sizer)  │   │ 最终下单规模 (Order)    │
├─────────────────────────┤   ├─────────────────────────┤   ├─────────────────────────┤
│ • 股票 A, B, C 同时开仓 │──►│ • 计算各标的 ATR 风险   │──►│ • 股票 A: 1200 股       │
│ • 生成原始买入请求      │   │ • 依据可用资金等比例缩放│   │ • 股票 B: 800 股        │
└─────────────────────────┘   └─────────────────────────┘   └─────────────────────────┘

一、头寸分配基类与固定比例风险模型面向对象实现

通过继承 Python abc.ABC，设计跨资产、跨策略复用的头寸分配基类：

$$\text{Size}_i = \min\left( \frac{\text{Capital} \times R_i}{\text{RiskPerShare}_i}, \frac{\text{Capital} \times w_{\max}}{P_i} \right)$$

from abc import ABC, abstractmethod
import numpy as np

class BasePositionSizer(ABC):
    """
    资金管理与头寸规模计算抽象基类
    """
    @abstractmethod
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        pass

class CPRPositionSizer(BasePositionSizer):
    """
    CPR 固定比例风险头寸分配器（合成教学样例，历史参数不代表未来）
    """
    def __init__(self, risk_pct: float = 0.015, atr_k: float = 2.5, max_weight: float = 0.20):
        self.risk_pct = risk_pct
        self.atr_k = atr_k
        self.max_weight = max_weight
        
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        risk_budget = total_equity * self.risk_pct
        risk_per_share = max(1e-4, atr * self.atr_k)
        
        # 1. 风险限制规模
        size_by_risk = int(risk_budget / risk_per_share)
        
        # 2. 最大单票持仓资金上限
        max_capital_for_stock = total_equity * self.max_weight
        size_by_capital = int(max_capital_for_stock / current_price)
        
        final_shares = min(size_by_risk, size_by_capital)
        return (final_shares // 100) * 100

# 合成数据演示：总权益 100 万，不同价格与波动率的股票
rng = np.random.default_rng(42)
total_equity = 1_000_000.0
price_samples = np.array([1800.0, 150.0, 220.0])
atr_samples = np.array([30.0, 4.0, 6.0])

sizer = CPRPositionSizer(risk_pct=0.015, atr_k=2.0, max_weight=0.15)
print("===== CPR 头寸分配器单资产计算演示 =====")
for i, (price, atr) in enumerate(zip(price_samples, atr_samples)):
    shares = sizer.calculate_size(total_equity, price, atr)
    cost = shares * price
    risk_per_share = max(1e-4, atr * sizer.atr_k)
    print(f"股票{i+1}: 价格 {price:7.2f}, ATR {atr:5.2f}, 风险/股 {risk_per_share:6.2f}")
    print(f"      分配股数 {shares:5d}, 占用资金 {cost:10.2f}, 仓位权重 {cost/total_equity*100:5.2f}%")

===== CPR 头寸分配器单资产计算演示 =====
股票1: 价格 1800.00, ATR 30.00, 风险/股  60.00
      分配股数     0, 占用资金       0.00, 仓位权重  0.00%
股票2: 价格  150.00, ATR  4.00, 风险/股   8.00
      分配股数  1000, 占用资金  150000.00, 仓位权重 15.00%
股票3: 价格  220.00, ATR  6.00, 风险/股  12.00
      分配股数   600, 占用资金  132000.00, 仓位权重 13.20%

二、多资产投资组合资金池动态配额管理

当多个标的在同一交易日产生买入信号时，头寸分配器必须进行全局预算协调，避免总头寸超调：

$$\sum_{i=1}^M \text{Cost}_i \le \text{Available Cash}$$

from abc import ABC, abstractmethod
import numpy as np

class BasePositionSizer(ABC):
    """
    资金管理与头寸规模计算抽象基类
    """
    @abstractmethod
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        pass

class CPRPositionSizer(BasePositionSizer):
    """
    CPR 固定比例风险头寸分配器（合成教学样例，历史参数不代表未来）
    """
    def __init__(self, risk_pct: float = 0.015, atr_k: float = 2.5, max_weight: float = 0.20):
        self.risk_pct = risk_pct
        self.atr_k = atr_k
        self.max_weight = max_weight
        
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        risk_budget = total_equity * self.risk_pct
        risk_per_share = max(1e-4, atr * self.atr_k)
        
        # 1. 风险限制规模
        size_by_risk = int(risk_budget / risk_per_share)
        
        # 2. 最大单票持仓资金上限
        max_capital_for_stock = total_equity * self.max_weight
        size_by_capital = int(max_capital_for_stock / current_price)
        
        final_shares = min(size_by_risk, size_by_capital)
        return (final_shares // 100) * 100

class MultiAssetPortfolioSizer:
    """
    多资产资金池配额管理器（合成教学样例，等比例缩放可能改变风险暴露）
    """
    def __init__(self, sizer: BasePositionSizer):
        self.sizer = sizer
        
    def allocate_multi_orders(self, total_equity: float, available_cash: float, signal_items: list) -> dict:
        orders = {}
        total_required_cash = 0
        price_map = {item['symbol']: item['price'] for item in signal_items}
        
        # 第一轮：按各自风险计算初选股数
        for item in signal_items:
            sym, price, atr = item['symbol'], item['price'], item['atr']
            shares = self.sizer.calculate_size(total_equity, price, atr)
            cost = shares * price
            orders[sym] = {'shares': shares, 'cost': cost}
            total_required_cash += cost
            
        # 第二轮：如果超预算，等比例缩放
        if total_required_cash > available_cash and total_required_cash > 0:
            scale = available_cash / total_required_cash
            for sym in orders:
                scaled_shares = int(orders[sym]['shares'] * scale)
                orders[sym]['shares'] = (scaled_shares // 100) * 100
                orders[sym]['cost'] = orders[sym]['shares'] * price_map[sym]
                
        return orders

# 合成多资产场景：总权益 100 万，可用现金 20 万，3 只股票同时触发
rng = np.random.default_rng(42)
total_equity = 1_000_000.0
available_cash = 200_000.0

signals = [
    {'symbol': '600519.SH', 'price': 1800.0, 'atr': 30.0},
    {'symbol': '000858.SZ', 'price': 150.0, 'atr': 4.0},
    {'symbol': '300750.SZ', 'price': 220.0, 'atr': 6.0}
]

sizer = CPRPositionSizer(risk_pct=0.015, atr_k=2.0, max_weight=0.15)
portfolio_manager = MultiAssetPortfolioSizer(sizer)
orders = portfolio_manager.allocate_multi_orders(total_equity, available_cash, signals)

print("===== 多资产资金池配额管理演示 =====")
print(f"总权益: {total_equity:.2f}, 可用现金: {available_cash:.2f}")
total_allocated = 0.0
for sym, order in sorted(orders.items()):
    print(f"标的 {sym:<10}: 分配 {order['shares']:>5} 股 | 占用资金: {order['cost']:>10.2f}")
    total_allocated += order['cost']
print(f"总占用资金: {total_allocated:.2f} (≤可用现金: {total_allocated <= available_cash})")

===== 多资产资金池配额管理演示 =====
总权益: 1000000.00, 可用现金: 200000.00
标的 000858.SZ : 分配   700 股 | 占用资金:  105000.00
标的 300750.SZ : 分配   400 股 | 占用资金:   88000.00
标的 600519.SH : 分配     0 股 | 占用资金:       0.00
总占用资金: 193000.00 (≤可用现金: True)

📝 动手练一练

编写 Python 脚本对 CPR 头寸分配器与多资产配额协调机制执行完整的单元测试：

import numpy as np

class BasePositionSizer:
    """
    资金管理与头寸规模计算基类（简化版，用于模块测试）
    """
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        raise NotImplementedError

class CPRPositionSizer(BasePositionSizer):
    """
    CPR 固定比例风险头寸分配器（合成教学样例，历史参数不代表未来）
    """
    def __init__(self, risk_pct: float = 0.015, atr_k: float = 2.5, max_weight: float = 0.20):
        self.risk_pct = risk_pct
        self.atr_k = atr_k
        self.max_weight = max_weight
        
    def calculate_size(self, total_equity: float, current_price: float, atr: float) -> int:
        risk_budget = total_equity * self.risk_pct
        risk_per_share = max(1e-4, atr * self.atr_k)
        size_by_risk = int(risk_budget / risk_per_share)
        max_capital_for_stock = total_equity * self.max_weight
        size_by_capital = int(max_capital_for_stock / current_price)
        final_shares = min(size_by_risk, size_by_capital)
        return (final_shares // 100) * 100

class MultiAssetPortfolioSizer:
    """
    多资产资金池配额管理器（合成教学样例，等比例缩放可能改变风险暴露）
    """
    def __init__(self, sizer: BasePositionSizer):
        self.sizer = sizer
        
    def allocate_multi_orders(self, total_equity: float, available_cash: float, signal_items: list) -> dict:
        orders = {}
        total_required_cash = 0
        price_map = {item['symbol']: item['price'] for item in signal_items}
        for item in signal_items:
            sym, price, atr = item['symbol'], item['price'], item['atr']
            shares = self.sizer.calculate_size(total_equity, price, atr)
            cost = shares * price
            orders[sym] = {'shares': shares, 'cost': cost}
            total_required_cash += cost
        if total_required_cash > available_cash and total_required_cash > 0:
            scale = available_cash / total_required_cash
            for sym in orders:
                scaled_shares = int(orders[sym]['shares'] * scale)
                orders[sym]['shares'] = (scaled_shares // 100) * 100
                orders[sym]['cost'] = orders[sym]['shares'] * price_map[sym]
        return orders

# 模块集成测试（合成数据，历史回测不代表未来收益）
rng = np.random.default_rng(42)
sizer = CPRPositionSizer(risk_pct=0.015, atr_k=2.0, max_weight=0.15)
portfolio_manager = MultiAssetPortfolioSizer(sizer)

equity = 1_000_000.0
cash = 200_000.0

signals = [
    {'symbol': '600519.SH', 'price': 1800.0, 'atr': 30.0},
    {'symbol': '000858.SZ', 'price': 150.0, 'atr': 4.0},
    {'symbol': '300750.SZ', 'price': 220.0, 'atr': 6.0}
]

res = portfolio_manager.allocate_multi_orders(equity, cash, signals)

print("================ 资金管理模块集成测试 ================")
total_allocated_cash = 0.0
for sym, order in sorted(res.items()):
    print(f"  - 标的 {sym:<10}: 分配 {order['shares']:>5} 股 | 占用资金: {order['cost']:>10.2f} 元")
    total_allocated_cash += order['cost']
print(f"总占用资金: {total_allocated_cash:.2f} 元 (可用现金上限: {cash:.2f} 元)")
budget_check = total_allocated_cash <= cash
print(f"预算控制: {'✅ PASS' if budget_check else '❌ FAIL'}")
print("====================================================")

运行结果输出：

================ 资金管理模块集成测试 ================
  - 标的 000858.SZ : 分配   700 股 | 占用资金:  105000.00 元
  - 标的 300750.SZ : 分配   400 股 | 占用资金:   88000.00 元
  - 标的 600519.SH : 分配     0 股 | 占用资金:       0.00 元
总占用资金: 193000.00 元 (可用现金上限: 200000.00 元)
预算控制: ✅ PASS
====================================================

本章小结

1. 面向对象接口解耦：采用抽象基类设计标准化资金管理接口，便于无缝切换凯利模型或风险平价模型。
2. 双重约束保障安全：同时施加单笔最大风险限额（Risk-based）与单票持仓上限（Capital-based）双重限制。
3. 支持多标的动态等比缩放：当全市场多信号并发时，自动依据账户实时购买力进行等比例头寸压缩。

📋 行动清单 (Action Checklist)

• 运行资金管理测试代码，测试在账户资金极端受限时的缩放表现；
• 编写基于波动率倒数加权（Inverse Volatility Sizer）的头寸分配类；
• 预习下一节，掌握将数据、信号、风控与资金管理组装为完整量化策略。

—— 小象教研组