实盘交易系统：执行算法与滑点控制

import pandas as pd
import numpy as np

def calculate_slippage(df_orders, df_trades):
    """
    计算滑点
    df_orders: 订单表 (timestamp, symbol, side, order_price, quantity)
    df_trades: 成交表 (timestamp, symbol, side, fill_price, quantity)
    """
    merged = pd.merge(df_orders, df_trades, on=['timestamp', 'symbol', 'side'])
    
    # 计算滑点（对买单，滑点 = order_price - fill_price）
    merged['slippage'] = np.where(
        merged['side'] == 'BUY',
        merged['order_price'] - merged['fill_price'],
        merged['fill_price'] - merged['order_price']
    )
    
    # 统计指标
    avg_slippage = merged['slippage'].mean()
    slippage_bps = avg_slippage / merged['order_price'].mean() * 10000  # 转换为基点
    
    return {
        'avg_slippage': avg_slippage,
        'slippage_bps': slippage_bps,
        'slippage_std': merged['slippage'].std(),
        'positive_rate': (merged['slippage'] > 0).mean()  # 正滑点占比
    }

# 示例输出
# {'avg_slippage': 0.0032, 'slippage_bps': 1.8, 'slippage_std': 0.0051, 'positive_rate': 0.12}

def vwap_schedule(total_shares, volume_profile, horizon_minutes=60):
    """
    VWAP 执行计划
    total_shares: 总股数
    volume_profile: 历史成交量分布 (每分钟)
    horizon_minutes: 执行时间窗口（分钟）
    """
    # 归一化成交量分布
    volume_profile = volume_profile[:horizon_minutes]
    volume_profile = volume_profile / volume_profile.sum()
    
    # 计算每分钟应执行的股数
    schedule = total_shares * volume_profile
    
    return schedule.cumsum()  # 累计执行计划

# 示例：3600 股分 60 分钟执行
# 如果前 10 分钟成交量占全天的 5%，则前 10 分钟执行 180 股

def is_optimal_schedule(total_shares, volatility, risk_aversion=1e-6):
    """
    Almgren-Chriss IS 模型
    volatility: 波动率
    risk_aversion: 风险厌恶系数（越大越保守）
    """
    # 模型参数
    eta = 0.05  # 临时市场冲击系数
    gamma = 0.1  # 永久市场冲击系数
    tau = 1.0  # 执行时间（归一化）
    
    # 最优执行轨迹（解析解）
    k = np.sqrt(gamma / (eta * risk_aversion * volatility**2))
    t = np.linspace(0, tau, 100)
    
    # 持仓变化
    x_t = total_shares * (np.cosh(k * (tau - t)) / np.cosh(k * tau))
    
    # 执行速率
    v_t = -np.gradient(x_t, t)
    
    return x_t, v_t

# 可视化最优执行轨迹
import matplotlib.pyplot as plt

x_t, v_t = is_optimal_schedule(10000, volatility=0.02)
plt.plot(x_t, label='持仓')
plt.plot(v_t, label='执行速率')
plt.legend()
plt.show()

┌─────────────────────────────────────────┐
│         策略层 (Strategy Layer)          │
│  - 产生交易信号                          │
│  - 计算目标持仓                          │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│       执行层 (Execution Layer)           │
│  - 订单拆分（VWAP/TWAP/IS）             │
│  - 滑点控制（限价单/隐藏订单）           │
│  - 智能路由（SOR）                       │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│       风控层 (Risk Layer)                │
│  - 仓位限制                              │
│  - 止损/止盈                             │
│  - 异常监控（熔断机制）                   │
└──────────────┬──────────────────────────┘
               │
┌──────────────▼──────────────────────────┐
│       交易接口 (Broker API)              │
│  - IB / Alpaca / TDAmeritrade           │
│  - 订单状态管理                          │
└─────────────────────────────────────────┘

def calculate_implementation_shortfall(df_backtest, df_live):
    """
    计算执行缺口
    """
    # 回测收益（假设无滑点）
    bt_return = df_backtest['equity'].pct_change().sum()
    
    # 实盘收益（含滑点）
    live_return = df_live['equity'].pct_change().sum()
    
    # 执行缺口
    is_cost = bt_return - live_return
    
    return {
        'backtest_return': bt_return,
        'live_return': live_return,
        'implementation_shortfall': is_cost,
        'is_bps': is_cost * 10000  # 转换为基点
    }

# 示例输出
# {'backtest_return': 0.15, 'live_return': 0.12, 'implementation_shortfall': 0.03, 'is_bps': 300}