03-交易成本模型
预计学习时间:1 小时
难度:⭐⭐⭐
核心问题:真实的交易成本有多高?如何在回测中准确建模?
为什么回测必须考虑成本?
一个震撼的例子
# 假设有一个高频策略
策略毛收益: +50%/年
换手率: 500%/年(平均持仓2天)
双边成本: 0.3%(佣金+印花税+滑点)
净收益 = 50% - 500% × 0.3% = 50% - 1.5% = 48.5%
# 还能接受
# 但如果成本估计不足:
真实双边成本: 0.8%(大额资金冲击成本更高)
净收益 = 50% - 500% × 0.8% = 50% - 4% = 46%
# 如果换手率更高:
换手率: 1000%/年
净收益 = 50% - 1000% × 0.3% = 50% - 3% = 47%结论:交易成本可以完全吞噬策略收益,甚至导致亏损!
成本对策略的影响
| 策略类型 | 典型换手率 | 成本敏感度 | 成本占比 |
|---|---|---|---|
| 低频(月度调仓) | 50%/年 | 低 | 0.15%/年 |
| 中频(周度调仓) | 200%/年 | 中 | 0.6%/年 |
| 高频(日度调仓) | 500%/年 | 高 | 1.5%/年 |
| 超高频(日内) | 5000%/年 | 极高 | 15%/年 |
关键认知:
如果不考虑交易成本:
├─ 高换手策略会显得异常优秀
├─ 参数优化会选择高换手参数
└─ 实盘会遭遇"收益消失"的灾难
成本构成
交易成本由多个部分组成,每一部分都需要准确建模:
┌────────────────────────────────────────────────────────────────┐
│ 交易成本构成 │
├────────────────────────────────────────────────────────────────┤
│ │
│ 总交易成本 │
│ ├─ 显性成本(可以看到的) │
│ │ ├─ 佣金/手续费 │
│ │ ├─ 印花税 │
│ │ └─ 过户费 │
│ │ │
│ └─ 隐性成本(不容易看到的) │
│ ├─ 滑点(Slippage) │
│ └─ 冲击成本(Market Impact) │
│ │
└────────────────────────────────────────────────────────────────┘
1. 佣金/手续费 (Commission)
A股常见费率
| 佣金类型 | 费率 | 说明 |
|---|---|---|
| 普通佣金 | 万2.5 - 万5 | 券商默认费率 |
| 佣金最低 | 5元 | 不足5元按5元收取 |
| VIP佣金 | 万1 - 万1.5 | 资金量大可协商 |
| 过户费 | 万0.1 | 仅沪市,买卖双向 |
代码实现
class CommissionModel:
"""佣金计算模型"""
def __init__(self, rate=0.0003, min_commission=5):
"""
参数
----
rate : float
佣金费率,默认万3
min_commission : float
最低佣金,默认5元
"""
self.rate = rate
self.min_commission = min_commission
def calculate(self, trade_amount):
"""
计算佣金
参数
----
trade_amount : float or Series
交易金额
返回
----
commission : float or Series
佣金金额
"""
commission = trade_amount * self.rate
commission = np.maximum(commission, self.min_commission)
return commission
# 示例
commission_model = CommissionModel(rate=0.0003, min_commission=5)
print("=== 佣金计算示例 ===")
trade_amounts = [100, 1000, 5000, 10000, 50000]
for amount in trade_amounts:
comm = commission_model.calculate(amount)
effective_rate = comm / amount if amount > 0 else 0
print(f"交易额: {amount:8.0f}元 → 佣金: {comm:5.2f}元 → 有效费率: {effective_rate:.4%}")输出:
=== 佣金计算示例 ===
交易额: 100元 → 佣金: 5.00元 → 有效费率: 5.0000%
交易额: 1000元 → 佣金: 5.00元 → 有效费率: 0.5000%
交易额: 5000元 → 佣金: 5.00元 → 有效费率: 0.1000%
交易额: 10000元 → 佣金: 5.00元 → 有效费率: 0.0500%
交易额: 50000元 → 佣金: 15.00元 → 有效费率: 0.0300%
注意:小额交易的有效费率极高!
阶梯费率
class TieredCommissionModel:
"""阶梯佣金模型"""
def __init__(self, tiers):
"""
参数
----
tiers : list of tuples
[(交易额下限, 费率), ...]
例如: [(0, 0.0003), (100000, 0.0002), (1000000, 0.0001)]
"""
self.tiers = sorted(tiers, key=lambda x: x[0])
def calculate(self, trade_amount):
"""计算阶梯佣金"""
rate = self.tiers[0][1]
for threshold, tier_rate in self.tiers:
if trade_amount >= threshold:
rate = tier_rate
return trade_amount * rate
# 示例
tiered = TieredCommissionModel([
(0, 0.0003), # 100万以下,万3
(1000000, 0.0002), # 100万-1000万,万2
(10000000, 0.0001) # 1000万以上,万1
])
print("\n=== 阶梯佣金 ===")
for amount in [500000, 2000000, 15000000]:
comm = tiered.calculate(amount)
print(f"交易额: {amount:10.0f}元 → 佣金: {comm:7.2f}元")2. 印花税 (Stamp Duty)
A股印花税规则
| 交易方向 | 费率 | 说明 |
|---|---|---|
| 买入 | 0 | 不收取 |
| 卖出 | 0.1% | 千分之一,仅卖出收取 |
| 基金/ETF | 0 | 免征印花税 |
⚠️ 重要:印花税是单边征收,只有卖出时才收取!
代码实现
class StampDutyModel:
"""印花税计算模型"""
def __init__(self, rate=0.001):
"""
参数
----
rate : float
印花税率,默认0.1%
"""
self.rate = rate
def calculate(self, trade_amount, is_sell=True):
"""
计算印花税
参数
----
trade_amount : float or Series
交易金额
is_sell : bool
是否为卖出交易
返回
----
stamp_duty : float or Series
印花税金额
"""
if is_sell:
return trade_amount * self.rate
else:
return 0
# 示例
stamp_duty = StampDutyModel(rate=0.001)
print("=== 印花税计算示例 ===")
trade_amount = 100000
buy_tax = stamp_duty.calculate(trade_amount, is_sell=False)
sell_tax = stamp_duty.calculate(trade_amount, is_sell=True)
print(f"买入 {trade_amount:,.0f}元: 印花税 {buy_tax:.2f}元")
print(f"卖出 {trade_amount:,.0f}元: 印花税 {sell_tax:.2f}元")
print(f"一轮买卖: 总印花税 {buy_tax + sell_tax:.2f}元")3. 滑点 (Slippage)
定义
滑点:预期成交价格与实际成交价格之间的差异。
滑点来源
- 买卖价差(Bid-Ask Spread):买入用卖一价,卖出用买一价
- 价格波动:下单到成交期间价格变动
- 流动性不足:大单需要分批成交,价格逐步恶化
滑点模型
线性滑点模型:
class LinearSlippageModel:
"""线性滑点模型"""
def __init__(self, rate=0.001):
"""
参数
----
rate : float
滑点率,默认0.1%
"""
self.rate = rate
def calculate(self, price, direction='buy'):
"""
计算滑点后的价格
参数
----
price : float or Series
预期成交价
direction : str
'buy' 或 'sell'
返回
----
execution_price : float or Series
实际成交价
"""
if direction == 'buy':
# 买入时价格更高
return price * (1 + self.rate)
else:
# 卖出时价格更低
return price * (1 - self.rate)
def calculate_cost(self, trade_value, direction='buy'):
"""
计算滑点造成的成本
参数
----
trade_value : float or Series
交易金额
direction : str
'buy' 或 'sell'
返回
----
slippage_cost : float or Series
滑点成本
"""
if direction == 'buy':
return trade_value * self.rate / (1 + self.rate)
else:
return trade_value * self.rate / (1 - self.rate)
# 示例
slippage = LinearSlippageModel(rate=0.001)
print("\n=== 滑点计算示例 ===")
prices = [10, 50, 100, 500]
for price in prices:
buy_price = slippage.calculate(price, 'buy')
sell_price = slippage.calculate(price, 'sell')
buy_cost = slippage.calculate_cost(100000, 'buy')
sell_cost = slippage.calculate_cost(100000, 'sell')
print(f"价格{price:4.0f}: 买{buy_price:.2f} 卖{sell_price:.2f} | "
f"滑点成本: 买{buy_cost:.2f} 卖{sell_cost:.2f}")基于成交量的滑点模型
更精确的模型考虑成交量:
class VolumeSlippageModel:
"""基于成交量的滑点模型"""
def __init__(self, volume_impact=0.1, base_rate=0.0005):
"""
参数
----
volume_impact : float
成交量影响系数
base_rate : float
基础滑点率
"""
self.volume_impact = volume_impact
self.base_rate = base_rate
def calculate(self, price, volume, trade_volume, direction='buy'):
"""
计算滑点
参数
----
price : float
当前价格
volume : float
日成交量
trade_volume : float
交易数量
direction : str
交易方向
返回
----
execution_price : float
实际成交价
"""
# 滑点率 = 基础滑点 + 成交量占比影响
volume_ratio = trade_volume / volume
slippage_rate = self.base_rate + self.volume_impact * volume_ratio
if direction == 'buy':
return price * (1 + slippage_rate)
else:
return price * (1 - slippage_rate)
# 示例
vol_slippage = VolumeSlippageModel()
print("\n=== 成交量滑点模型示例 ===")
price = 100
daily_volume = 1000000 # 日成交100万股
trade_volumes = [1000, 10000, 50000, 100000] # 不同的交易量
for tv in trade_volumes:
buy_price = vol_slippage.calculate(price, daily_volume, tv, 'buy')
sell_price = vol_slippage.calculate(price, daily_volume, tv, 'sell')
print(f"交易{tv:7.0f}股 (占{tv/daily_volume:.2%}): "
f"买{buy_price:.2f} 卖{sell_price:.2f}")4. 冲击成本 (Market Impact)
定义
冲击成本:大额交易推动价格朝不利方向变动的成本。
当你的交易量占市场成交量比例较大时,你的买卖会推高/压低价格。
Almgren-Chriss 模型(简化版)
class MarketImpactModel:
"""市场冲击成本模型"""
def __init__(self, impact_coef=0.5, vol_adjust=True):
"""
参数
----
impact_coef : float
冲击系数
vol_adjust : bool
是否根据波动率调整
"""
self.impact_coef = impact_coef
self.vol_adjust = vol_adjust
def calculate(self, price, volume, trade_volume, volatility=None):
"""
计算冲击成本
参数
----
price : float
当前价格
volume : float
日成交量(金额)
trade_volume : float
交易金额
volatility : float, optional
日波动率
返回
----
impact_cost : float
冲击成本(占交易金额的比例)
"""
# 基础冲击
volume_ratio = trade_volume / volume
impact = self.impact_coef * volume_ratio
# 波动率调整
if self.vol_adjust and volatility is not None:
# 年化波动率转为日波动率
daily_vol = volatility / np.sqrt(252)
impact *= daily_vol / 0.02 # 相对于2%日波动率
return impact
# 示例
impact_model = MarketImpactModel(impact_coef=0.5)
print("\n=== 冲击成本计算示例 ===")
price = 100
shares = 1000000 # 日成交100万股
volume = price * shares
volatility = 0.3 # 30%年化波动率
trade_shares = [10000, 50000, 100000, 200000]
for ts in trade_shares:
trade_value = price * ts
impact = impact_model.calculate(price, volume, trade_value, volatility)
print(f"交易{ts:7.0f}股 ({ts/shares:.2%}): "
f"冲击成本 {impact:.4%} = {trade_value * impact:8.2f}元")完整成本计算代码
TransactionCost 类
import numpy as np
import pandas as pd
class TransactionCost:
"""完整交易成本计算"""
def __init__(self,
commission_rate=0.0003,
min_commission=5,
stamp_duty_rate=0.001,
slippage_rate=0.001,
impact_coef=0.5):
"""
参数
----
commission_rate : float
佣金费率
min_commission : float
最低佣金
stamp_duty_rate : float
印花税率
slippage_rate : float
滑点率
impact_coef : float
冲击系数
"""
self.commission_rate = commission_rate
self.min_commission = min_commission
self.stamp_duty_rate = stamp_duty_rate
self.slippage_rate = slippage_rate
self.impact_coef = impact_coef
def calculate_trade_cost(self, price, shares, direction='buy',
volume=None, volatility=None):
"""
计算单笔交易的总成本
参数
----
price : float
成交价格
shares : float
成交数量
direction : str
'buy' 或 'sell'
volume : float, optional
日成交量(用于计算冲击)
volatility : float, optional
年化波动率
返回
----
cost_breakdown : dict
各项成本的明细
"""
trade_value = price * shares
# 1. 佣金
commission = trade_value * self.commission_rate
commission = max(commission, self.min_commission)
# 2. 印花税
stamp_duty = trade_value * self.stamp_duty_rate if direction == 'sell' else 0
# 3. 滑点成本
slippage_cost = trade_value * self.slippage_rate
# 4. 冲击成本
impact_cost = 0
if volume is not None:
volume_ratio = trade_value / volume
impact_cost = trade_value * self.impact_coef * volume_ratio
if volatility is not None:
daily_vol = volatility / np.sqrt(252)
impact_cost *= daily_vol / 0.02
total_cost = commission + stamp_duty + slippage_cost + impact_cost
return {
'trade_value': trade_value,
'commission': commission,
'stamp_duty': stamp_duty,
'slippage': slippage_cost,
'market_impact': impact_cost,
'total': total_cost,
'rate': total_cost / trade_value if trade_value > 0 else 0
}
def calculate_portfolio_turnover(self, old_weights, new_weights, portfolio_value):
"""
计算组合换手率和总成本
参数
----
old_weights : Series
调仓前权重
new_weights : Series
目标权重
portfolio_value : float
组合总价值
返回
----
result : dict
包含换手率和成本信息
"""
# 计算权重变化
weight_change = new_weights - old_weights
# 买入和卖出
buy_value = (weight_change[weight_change > 0]).sum() * portfolio_value
sell_value = (-weight_change[weight_change < 0]).sum() * portfolio_value
# 换手率(单边)
turnover = (abs(weight_change).sum() / 2)
# 估算成本(假设平均价格和成交量)
avg_buy_cost = buy_value * (self.commission_rate + self.slippage_rate)
avg_sell_cost = sell_value * (self.commission_rate + self.slippage_rate +
self.stamp_duty_rate)
total_cost = avg_buy_cost + avg_sell_cost
return {
'buy_value': buy_value,
'sell_value': sell_value,
'turnover': turnover,
'buy_cost': avg_buy_cost,
'sell_cost': avg_sell_cost,
'total_cost': total_cost,
'cost_rate': total_cost / portfolio_value if portfolio_value > 0 else 0
}
# 示例
cost_model = TransactionCost()
print("=" * 60)
print("完整的交易成本计算")
print("=" * 60)
# 单笔交易成本
cost_breakdown = cost_model.calculate_trade_cost(
price=100,
shares=1000,
direction='buy',
volume=10000000, # 日成交1000万
volatility=0.3
)
print("\n=== 单笔买入 1000 股 @ 100元 ===")
for key, value in cost_breakdown.items():
if key != 'rate':
print(f"{key:15s}: {value:10.2f}元")
else:
print(f"{key:15s}: {value:10.4%}")
print("\n=== 单笔卖出 1000 股 @ 100元 ===")
cost_breakdown = cost_model.calculate_trade_cost(
price=100,
shares=1000,
direction='sell',
volume=10000000,
volatility=0.3
)
for key, value in cost_breakdown.items():
if key != 'rate':
print(f"{key:15s}: {value:10.2f}元")
else:
print(f"{key:15s}: {value:10.4%}")
# 组合换手率成本
old_weights = pd.Series([0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1],
index=[f'S{i}' for i in range(10)])
new_weights = pd.Series([0.15, 0.05, 0.12, 0.08, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1],
index=[f'S{i}' for i in range(10)])
turnover_result = cost_model.calculate_portfolio_turnover(
old_weights, new_weights, portfolio_value=1000000
)
print("\n=== 组合换手率成本 (组合价值100万) ===")
print(f"买入金额: {turnover_result['buy_value']:,.2f}元")
print(f"卖出金额: {turnover_result['sell_value']:,.2f}元")
print(f"换手率: {turnover_result['turnover']:.2%}")
print(f"总成本: {turnover_result['total_cost']:,.2f}元")
print(f"成本率: {turnover_result['cost_rate']:.4%}")成本对收益的影响分析
不同换手率下的收益对比
def analyze_cost_impact():
"""分析成本对不同换手率策略的影响"""
cost_model = TransactionCost()
# 假设策略毛收益
gross_return = 0.20 # 20%年化
# 不同换手率
turnovers = [0.2, 0.5, 1.0, 2.0, 5.0, 10.0] # 20% - 1000%
results = []
for turnover in turnovers:
# 估算年化成本
# 双边成本 = 佣金 + 印花税(单边) + 滑点
bid_ask_cost = (cost_model.commission_rate +
cost_model.slippage_rate * 2 +
cost_model.stamp_duty_rate / 2)
annual_cost = turnover * bid_ask_cost
net_return = gross_return - annual_cost
results.append({
'换手率': f"{turnover:.0%}",
'年化成本': f"{annual_cost:.2%}",
'净收益': f"{net_return:.2%}",
'成本占比': f"{annual_cost/gross_return:.1%}"
})
df = pd.DataFrame(results)
print("\n=== 成本对不同换手率策略的影响 ===")
print(df.to_string(index=False))
analyze_cost_impact()输出示例:
=== 成本对不同换手率策略的影响 ===
换手率 年化成本 净收益 成本占比
20% 0.10% 19.90% 0.5%
50% 0.26% 19.74% 1.3%
100% 0.52% 19.48% 2.6%
200% 1.04% 18.96% 5.2%
500% 2.60% 17.40% 13.0%
1000% 5.20% 14.80% 26.0%
降低交易成本的方法
| 方法 | 说明 | 效果 |
|---|---|---|
| 降低调仓频率 | 周调→月调 | 换手率降低50%-75% |
| 设置阈值 | 权重变化<5%不调仓 | 换手率降低30%-50% |
| 选择流动性好的股票 | 过滤小盘股 | 降低冲击成本 |
| 分批交易 | 大单拆分 | 降低冲击成本 |
| 算法交易 | VWAP/TWAP | 降低冲击成本 |
T+1 限制
A股T+1规则
┌────────────────────────────────────────────────────────────────┐
│ A股T+1交易规则 │
├────────────────────────────────────────────────────────────────┤
│ │
│ 今天买入的股票 ─────────────────────────→ 明天才能卖出 │
│ │
│ 影响: │
│ ├─ 调仓信号当天只能执行买入部分 │
│ ├─ 卖出部分只能用之前持有的股票 │
│ └─ 可能导致持仓偏离目标 │
│ │
└────────────────────────────────────────────────────────────────┘
代码实现
def apply_t1_constraint(target_weights, current_weights):
"""
应用T+1限制
参数
----
target_weights : Series
目标权重(考虑T+1后的目标)
current_weights : Series
当前持仓权重
返回
----
executable_weights : Series
考虑T+1后实际能执行的权重
"""
# 今天买入的股票,明天才能卖出
# 所以今天的卖出只能用之前持有的股票
executable = current_weights.copy()
# 可以买入任意股票
executable = target_weights.copy()
# 但今天新买入的股票不能今天卖出
# 这里简化处理:实际回测中需要跟踪持仓历史
return executable
# 回测中的T+1处理
def backtest_with_t1(prices, signals, rebalance_freq='weekly'):
"""
考虑T+1的回测
在实际回测中:
1. 每个调仓日,根据前一天的信号生成目标权重
2. 今天只能卖出之前持有的股票
3. 今天买入的股票明天才能卖出
"""
# 简化示例
pass核心知识点总结
┌────────────────────────────────────────────────────────────────┐
│ 03-交易成本模型 核心要点 │
├────────────────────────────────────────────────────────────────┤
│ │
│ 1. 成本构成 │
│ ├─ 显性成本:佣金、印花税、过户费 │
│ └─ 隐性成本:滑点、冲击成本 │
│ │
│ 2. A股特殊规则 │
│ ├─ 印花税单边征收(卖出0.1%) │
│ ├─ T+1限制(当天买次日才能卖) │
│ └─ 佣金最低5元 │
│ │
│ 3. 成本影响 │
│ ├─ 高换手策略成本占比可达20%+ │
│ ├─ 成本可能完全吞噬Alpha │
│ └─ 必须在回测中准确建模 │
│ │
│ 4. 降低成本的方法 │
│ ├─ 降低调仓频率 │
│ ├─ 设置权重变动阈值 │
│ ├─ 选择高流动性股票 │
│ └─ 分批交易/算法交易 │
│ │
└────────────────────────────────────────────────────────────────┘
思考题
- 为什么同样的策略,资金规模越大,收益率越低?
- 一个换手率500%的策略,至少需要多少毛收益才能盈利?
- 如何在回测中准确估计冲击成本?
下一步
现在你知道如何建模交易成本了。接下来,让我们学习如何评估策略的表现。