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

用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 pandas as pd
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

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

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

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

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

📝 动手练一练

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

import pandas as pd
import numpy as np

def run_sizer_module_test():
    np.random.seed(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
    
    # 模拟 3 只股票同时触发买入
    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("================ 资金管理模块 Python 实现测试 ================")
    total_allocated_cash = 0
    for sym, order in 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} 元)")
    assert total_allocated_cash <= cash, "资金管理分配超出可用现金！"
    print("✅ 资金管理模块面向对象架构与预算控制 PASS")
    print("============================================================")

if __name__ == '__main__':
    run_sizer_module_test()

运行结果输出：

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

本章小结

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

📋 行动清单 (Action Checklist)

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

— 小象教研组