智能债券久期计算系统开发实战

class BondDurationSystem:
    def __init__(self):
        self.data_engine = MarketDataFetcher()  # 市场数据获取 
        self.calculator = DurationCalculator()  # 久期计算核心 
        self.visualizer = RiskVisualizer()     # 风险可视化 
        
    def analyze_bond(self, bond_code):
        # 全流程分析
        raw_data = self.data_engine.fetch(bond_code)
        cash_flows = self._process_cashflows(raw_data)
        results = self.calculator.compute(cash_flows)
        return self.visualizer.generate_report(results)

def build_cashflows(bond_info):
    """构建债券现金流时间表"""
    issue_date = bond_info['起息日']
    maturity = bond_info['期限']
    coupon_rate = bond_info['票面利率']
    
    # 生成付息时间序列
    dates = pd.date_range(issue_date, periods=maturity*2, freq='6M')  # 半年付息
    cashflows = [{'date': d, 'amount': 100*coupon_rate/2} for d in dates[:-1]]
    cashflows.append({'date': dates[-1], 'amount': 100*(1+coupon_rate/2)})  # 末次付息还本
    return pd.DataFrame(cashflows)

class DurationCalculator:
    def macaulay_duration(self, cashflows, ytm):
        """麦考利久期计算 """
        present_values = [cf['amount'] / (1+ytm)**(i/2) 
                         for i, cf in enumerate(cashflows.itertuples())]
        weighted_times = [pv * (cf.Index/2) for pv, cf in zip(present_values, cashflows.itertuples())]
        return sum(weighted_times) / sum(present_values)
    
    def modified_duration(self, mac_duration, ytm):
        """修正久期计算 """
        return mac_duration / (1 + ytm/2)

import akshare as ak

def fetch_yield_curve():
    """获取国开债实时收益率 """
    df = ak.bond_china_close_return()
    curve_data = df[df['债券简称'].str.contains('国开')]
    return curve_data[['期限', '到期收益率']].set_index('期限')

from scipy.interpolate import CubicSpline

class YieldCurveInterpolator:
    def __init__(self, curve_data):
        self.tenors = curve_data.index.values
        self.yields = curve_data['到期收益率'].values
        self.spline = CubicSpline(self.tenors, self.yields)  # 三次样条插值 
    
    def get_rate(self, t):
        return self.spline(t)

import plotly.express as px

def plot_risk_surface(duration, convexity):
    """绘制利率风险三维曲面 """
    x = np.linspace(-2, 2, 50)  # 利率变动幅度（%）
    price_changes = -duration*x + 0.5*convexity*(x**2)
    
    fig = px.scatter_3d(x=x, y=np.zeros_like(x), z=price_changes,
                        labels={'x':'利率变动', 'z':'价格波动'},
                        title="久期-凸性风险曲面")
    fig.update_traces(marker=dict(size=4, color=price_changes))
    return fig

def stress_test(bond_data, shock_scenarios):
    """利率冲击压力测试 """
    results = []
    for scenario in shock_scenarios:
        new_ytm = bond_data['ytm'] + scenario/100
        new_duration = self.calculator.modified_duration(
            bond_data['macaulay'], new_ytm)
        delta_price = -new_duration * scenario
        results.append({'冲击幅度': scenario, '久期': new_duration, 
                      '价格波动': delta_price})
    return pd.DataFrame(results)

def portfolio_duration(holdings):
    """组合久期计算 """
    total_value = sum(bond['market_value'] for bond in holdings)
    weighted_durations = [bond['duration'] * bond['market_value']/total_value 
                        for bond in holdings]
    return sum(weighted_durations)

# 示例组合
portfolio = [
    {'code': '220205.IB', 'market_value': 5e6, 'duration': 4.2},
    {'code': '210207.IB', 'market_value': 3e6, 'duration': 6.8}
]
print(f"组合久期：{portfolio_duration(portfolio):.2f}年")

def hedge_ratio(target_duration, current_duration, futures_duration):
    """计算套期保值比率 """
    return (target_duration - current_duration) / futures_duration