python实现 rebalance-杰瑞科技汇

Rebalance（再平衡）本身不是一个具体的 Python 函数，而是一种投资策略或算法思想，它的核心思想是：定期或不定期地调整投资组合中各类资产的比例，使其恢复到预设的“目标配置”（Target Allocation）。

（图片来源网络，侵删）

为什么需要再平衡？

想象一个简单的投资组合：50% 的股票基金和 50% 的债券基金。

一年后：股票大涨了30%，而债券只涨了5%。
新的比例：组合总价值增长，但股票的占比可能上升到了 65%，而债券下降到了 35%。

这偏离了你最初“50/50”的风险收益目标，股票占比过高，意味着组合的风险也变高了。再平衡 就派上用场了：

卖出部分股票：把股票的占比从 65% 卖掉，使其回到 50%。
买入债券：用卖出股票的钱买入债券，使其占比从 35% 提升到 50%。

这样做的好处是：

风险控制：防止单一资产类别风险过高。
“低买高卖”：在高估的资产（股票）上获利，在低估的资产（债券）上买入，这是一种纪律性的投资行为。
纪律性：避免因市场情绪而做出非理性决策。

Python 实现：一个完整的再平衡示例

下面,我将用一个完整的 Python 类来模拟和执行再平衡操作，这个例子将包含：

（图片来源网络，侵删）

定义目标配置
模拟市场波动导致资产比例变化
执行再平衡逻辑
展示再平衡前后的对比

第1步：定义资产和跟踪其价值

我们首先需要一个数据结构来表示我们的投资组合,一个简单的字典就足够了，键是资产名称，值是该资产的当前市值。

第2步：实现再平衡逻辑

再平衡的核心计算是：

计算当前总资产。
计算每个资产在目标配置下应有的市值 (目标总资产 * 目标占比)。
计算每个资产需要买卖的金额 (应有市值 - 当前市值)。
- 正数代表需要买入。
- 负数代表需要卖出。

第3步：编写完整的 Python 代码

import pprint
class PortfolioRebalancer:
    """
    一个简单的投资组合再平衡器。
    """
    def __init__(self, target_allocation: dict, initial_values: dict):
        """
        初始化投资组合。
        :param target_allocation: 目标配置字典，如 {'Stocks': 0.6, 'Bonds': 0.4}
        :param initial_values: 初始资产市值字典，如 {'Stocks': 10000, 'Bonds': 10000}
        """
        if not all(abs(sum(target_allocation.values()) - 1.0) < 1e-9 for _ in [target_allocation]):
            raise ValueError("目标配置的比例总和必须等于 1.0 或 100%。")
        self.target_allocation = target_allocation
        self.current_values = initial_values
        self.history = [] # 用于记录每次再平衡后的状态
    def get_total_value(self) -> float:
        """计算当前投资组合的总价值。"""
        return sum(self.current_values.values())
    def get_current_allocation(self) -> dict:
        """计算当前资产配置的比例。"""
        total = self.get_total_value()
        if total == 0:
            return {asset: 0.0 for asset in self.current_values}
        return {asset: (value / total) for asset, value in self.current_values.items()}
    def simulate_market_change(self, changes: dict):
        """
        模拟市场波动，改变资产价值。
        这是一个辅助函数，用于演示再平衡的必要性。
        :param changes: 一个字典，键是资产，值是该资产价值的百分比变化（0.3 表示 +30%）。
        """
        print("\n--- 模拟市场波动 ---")
        for asset, change_percent in changes.items():
            if asset not in self.current_values:
                raise ValueError(f"资产 '{asset}' 不在投资组合中。")
            old_value = self.current_values[asset]
            new_value = old_value * (1 + change_percent)
            self.current_values[asset] = new_value
            print(f"{asset} 价值变化: {old_value:.2f} -> {new_value:.2f} (变化率: {change_percent*100:.1f}%)")
    def rebalance(self, threshold: float = 0.05) -> dict:
        """
        执行再平衡操作。
        :param threshold: 再平衡阈值，只有当某个资产的当前配置与目标配置的差异
                          超过这个阈值时，才会触发对该资产的再平衡。
                          (0.05 表示 5% 的差异)。
        :return: 一个字典，包含需要买入或卖出的资产和金额。
        """
        print("\n--- 开始再平衡 ---")
        total_value = self.get_total_value()
        current_allocation = self.get_current_allocation()
        # 计算每个资产的应有价值和交易量
        trades = {}
        for asset in self.target_allocation:
            target_value = total_value * self.target_allocation[asset]
            current_value = self.current_values[asset]
            # 计算当前配置与目标配置的差异
            allocation_diff = abs(current_allocation.get(asset, 0) - self.target_allocation[asset])
            # 只有当差异超过阈值时才进行交易
            if allocation_diff > threshold:
                trade_amount = target_value - current_value
                trades[asset] = trade_amount
        if not trades:
            print("当前配置与目标配置的差异在阈值内，无需再平衡。")
            return {}
        # 执行交易（这里只是逻辑上的，实际应用中会调用券商API）
        print("执行交易...")
        for asset, amount in trades.items():
            self.current_values[asset] += amount
            action = "买入" if amount > 0 else "卖出"
            print(f"  {action} {asset}: {abs(amount):.2f}")
        # 记录再平衡后的状态
        self.history.append({
            'total_value': total_value,
            'current_allocation': self.get_current_allocation(),
            'trades': trades
        })
        return trades
    def print_status(self, title: str):
        """打印投资组合的当前状态。"""
        print(f"\n--- {title} ---")
        total_value = self.get_total_value()
        current_allocation = self.get_current_allocation()
        print(f"总资产价值: {total_value:,.2f}")
        print("当前配置:")
        pprint.pprint(current_allocation)
        print("目标配置:")
        pprint.pprint(self.target_allocation)
# --- 主程序 ---
if __name__ == "__main__":
    # 1. 定义目标配置和初始投资组合
    # 目标是60%的股票，40%的债券
    target_alloc = {'Stocks': 0.6, 'Bonds': 0.4}
    # 初始投资1万元股票，1万元债券
    initial_values = {'Stocks': 10000, 'Bonds': 10000}
    # 2. 创建再平衡器实例
    rebalancer = PortfolioRebalancer(target_allocation=target_alloc, initial_values=initial_values)
    # 3. 打印初始状态
    rebalancer.print_status("初始投资组合状态")
    # 4. 模拟市场波动（股票大涨，债券微涨）
    market_changes = {'Stocks': 0.30, 'Bonds': 0.05} # 股票涨30%，债券涨5%
    rebalancer.simulate_market_change(market_changes)
    # 5. 打印市场波动后的状态
    rebalancer.print_status("市场波动后的投资组合状态")
    # 6. 执行再平衡
    # 设置阈值为5%，即只有当配置偏离超过5%时才交易
    rebalancer.rebalance(threshold=0.05)
    # 7. 打印再平衡后的最终状态
    rebalancer.print_status("再平衡后的投资组合状态")
    # (可选) 模拟下一次市场波动和再平衡
    print("\n" + "="*50)
    print("模拟下一年的投资和再平衡...")
    next_market_changes = {'Stocks': -0.10, 'Bonds': 0.08} # 股票跌10%，债券涨8%
    rebalancer.simulate_market_change(next_market_changes)
    rebalancer.print_status("下一年市场波动后的状态")
    rebalancer.rebalance(threshold=0.05)
    rebalancer.print_status("最终再平衡后的状态")

代码解释

PortfolioRebalancer 类：封装了所有再平衡相关的逻辑。
__init__：初始化目标配置和当前资产价值，这里做了一个简单的检查，确保目标配置的比例加起来是100%。
get_total_value() 和 get_current_allocation()：是辅助方法，用于获取组合的总价值和当前各类资产的占比。
simulate_market_change(changes)：这是一个非常关键的演示函数，在真实世界中，市场数据会来自数据源（如 yfinance 库），这里我们用一个字典来手动模拟价格变动，从而改变 self.current_values，以便触发再平衡。
rebalance(threshold)：这是核心的再平衡函数。
- 它首先计算当前的总价值和各类资产的应有价值。
- 它引入了 threshold（阈值） 的概念，这是一个非常重要的实践，因为频繁的再平衡会产生交易成本，只有当配置偏离目标足够大时（比如超过5%），才进行交易。
- 它计算出每个资产的交易量,并更新 self.current_values。
- 它返回一个交易指令字典,方便后续处理（连接真实的券商API下单）。
print_status()：用于格式化打印投资组合的清晰状态，方便我们观察结果。
if __name__ == "__main__":：展示了如何使用这个类，从创建实例，到模拟市场变化，再到执行再平衡，整个过程非常清晰。

运行结果分析

当你运行上面的代码时,你会看到类似下面的输出：

--- 初始投资组合状态 ---
总资产价值: 20,000.00
当前配置:
{'Bonds': 0.5, 'Stocks': 0.5}
目标配置:
{'Bonds': 0.4, 'Stocks': 0.6}
--- 模拟市场波动 ---
Stocks 价值变化: 10000.00 -> 13000.00 (变化率: 30.0%)
Bonds 价值变化: 10000.00 -> 10500.00 (变化率: 5.0%)
--- 市场波动后的投资组合状态 ---
总资产价值: 23,500.00
当前配置:
{'Bonds': 0.44680851063829785, 'Stocks': 0.5531914893617022}
目标配置:
{'Bonds': 0.4, 'Stocks': 0.6}
--- 开始再平衡 ---
执行交易...
  卖出 Stocks: -925.00
--- 再平衡后的投资组合状态 ---
总资产价值: 23,500.00
当前配置:
{'Bonds': 0.4, 'Stocks': 0.6}
目标配置:
{'Bonds': 0.4, 'Stocks': 0.6}

从结果中可以清楚地看到：

（图片来源网络，侵删）

初始配置是50/50。
市场波动后,股票占比上升到了约55.3%，债券下降到了约44.7%。
再平衡逻辑被触发,计算出需要卖出价值925元的股票。
交易执行后,组合的配置完美地恢复到了目标值60/40。

扩展和实际应用

这个示例是教学性质的,在实际的量化交易或个人理财应用中，你可以在此基础上进行扩展：

从API获取数据：使用 yfinance、akshare 或券商提供的API来获取实时的资产价格。
自动化交易：将 rebalance 方法中的 trades 字典连接到券商的交易API（如富途牛牛、雪球的API等），实现自动下单。
更复杂的再平衡策略：
- 时间触发：除了阈值触发，还可以固定时间触发（如每个季度末）。
- 动态目标配置：目标配置本身也可以根据市场环境或投资者的年龄变化而调整（生命周期基金）。
成本分析：在计算交易时，加入交易手续费、印花税等成本，使模型更贴近现实。
回测：使用历史数据来测试你的再平衡策略在过去的表现如何。

python实现 rebalance

为什么需要再平衡？