aszer/etf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(execution): 回测调仓事件记录功能增强

Browse Source 
新增调仓事件记录功能，详细记录每次调仓的信息：

核心改进：
1. BacktestExecutor新增_apply_trade_cost_with_events方法
   - 记录每次调仓的基本信息（持仓变化、调入调出标的）
   - 记录换手率、调仓成本、持仓天数、当日收益

2. 新增_enrich_rebalance_events方法
   - 补充净值信息（调仓前净值、调仓后净值、净值变化%）

3. strategy.py保存调仓记录到CSV
   - 新增rebalances.csv文件
   - 返回结果包含rebalance_events

调仓记录字段：
- 调仓前持仓、调仓后持仓
- 调入标的、调出标的
- 换手率、调仓成本
- 持仓天数、当日收益
- 调仓前净值、调仓后净值、净值变化%

应用场景：
- 分析每次调仓对收益的影响
- 评估调仓决策质量
- 统计调仓频率与效果
This commit is contained in:
aszerW
2026-05-16 21:15:31 +08:00
parent 6308627f8c
commit 63c56f0001
2 changed files with 178 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
framework/execution/__init__.py
View File

@@ -193,8 +193,8 @@ class BacktestExecutor(Executor):
# 计算策略日收益率
result = self._calculate_daily_returns(signals, data, signal_col)
# 扣除交易成本
result = self._apply_trade_cost(result, signals, signal_col)
# 扣除交易成本（同时记录调仓事件）
result, rebalance_events = self._apply_trade_cost_with_events(result, signals, signal_col)
# 计算净值（起点归一化）
result = self._calculate_net_value(result)
@@ -208,6 +208,12 @@ class BacktestExecutor(Executor):
# 存储回测结果
portfolio.backtest_result = result
portfolio.rebalance_events = rebalance_events # 新增：调仓事件记录
# 补充调仓事件的净值信息
if not rebalance_events.empty:
rebalance_events = self._enrich_rebalance_events(rebalance_events, result)
portfolio.rebalance_events = rebalance_events
return portfolio
@@ -274,6 +280,162 @@ class BacktestExecutor(Executor):
return result
def _apply_trade_cost_with_events(self, result: pd.DataFrame, signals: pd.DataFrame, signal_col: str = 'signal') -> tuple:
"""
扣除交易成本并记录调仓事件
Returns:
(result, rebalance_events): 回测结果DataFrame和调仓事件DataFrame
"""
prev_signal = signals[signal_col].shift(1)
# 记录调仓事件
rebalance_events = []
last_rebalance_date = None
# 先计算累积收益率（用于计算调仓前后的净值）
cum_return_before_cost = result['策略日收益率'].copy()
if self.select_num == 1:
# 单标的策略
for i, (date, curr, prev) in enumerate(zip(signals.index, signals[signal_col], prev_signal)):
# 检查是否调仓
is_rebalance = False
turnover = 0.0
added = []
removed = []
if pd.notna(prev) and curr != prev:
is_rebalance = True
turnover = 1.0 if prev else 0.0
added = [curr] if curr else []
removed = [prev] if prev else []
# 扣除成本
result.loc[date, '策略日收益率'] -= self.trade_cost
# 记录调仓事件
if is_rebalance:
# 计算持仓天数
holding_days = 0
if last_rebalance_date is not None:
holding_days = (date - last_rebalance_date).days
event = {
'日期': date,
'调仓前持仓': prev if pd.notna(prev) else '',
'调仓后持仓': curr,
'调入标的': ','.join(added) if added else '',
'调出标的': ','.join(removed) if removed else '',
'换手率': turnover,
'调仓成本': self.trade_cost * turnover,
'持仓天数': holding_days,
'当日收益': result.loc[date, '策略日收益率'] + self.trade_cost * turnover, # 原始收益（扣除成本前）
}
rebalance_events.append(event)
last_rebalance_date = date
else:
# 多标的策略
turnover_list = []
for i, (date, curr, prev) in enumerate(zip(signals.index, signals[signal_col], prev_signal)):
# 检查是否调仓
is_rebalance = False
turnover = 0.0
added = []
removed = []
if pd.notna(prev) and curr != prev:
old = set(prev.split(',')) if prev else set()
new = set(curr.split(',')) if curr else set()
added = list(new - old)
removed = list(old - new)
swapped = len(removed)
turnover = swapped / len(old) if old else 0.0
is_rebalance = len(added) > 0 or len(removed) > 0
turnover_list.append(turnover)
# 扣除成本
result.loc[date, '策略日收益率'] -= turnover * self.trade_cost
else:
turnover_list.append(0.0)
# 记录调仓事件
if is_rebalance:
# 计算持仓天数
holding_days = 0
if last_rebalance_date is not None:
holding_days = (date - last_rebalance_date).days
event = {
'日期': date,
'调仓前持仓': prev if pd.notna(prev) else '',
'调仓后持仓': curr,
'调入标的': ','.join(added) if added else '',
'调出标的': ','.join(removed) if removed else '',
'换手率': turnover,
'调仓成本': self.trade_cost * turnover,
'持仓天数': holding_days,
'当日收益': result.loc[date, '策略日收益率'] + turnover * self.trade_cost, # 原始收益（扣除成本前）
}
rebalance_events.append(event)
last_rebalance_date = date
result['换手率'] = turnover_list
# 转换为DataFrame
rebalance_df = pd.DataFrame(rebalance_events) if rebalance_events else pd.DataFrame()
if not rebalance_df.empty:
rebalance_df['日期'] = pd.to_datetime(rebalance_df['日期'])
rebalance_df = rebalance_df.set_index('日期')
return result, rebalance_df
def _enrich_rebalance_events(self, rebalance_df: pd.DataFrame, result: pd.DataFrame) -> pd.DataFrame:
"""
补充调仓事件的净值信息
Args:
rebalance_df: 调仓事件DataFrame
result: 回测结果DataFrame（含净值序列）
Returns:
补充净值信息后的调仓事件DataFrame
"""
# 计算调仓前后净值变化
nav_before_list = []
nav_after_list = []
nav_change_list = []
for date in rebalance_df.index:
# 获取调仓日的净值
if date in result.index:
# 调仓前净值：前一天收盘净值
prev_date_idx = result.index.get_loc(date) - 1
if prev_date_idx >= 0:
nav_before = result['策略净值'].iloc[prev_date_idx]
else:
nav_before = 1.0
# 调仓后净值：当天收盘净值
nav_after = result.loc[date, '策略净值']
# 净值变化
nav_change = (nav_after / nav_before - 1) * 100
else:
nav_before = None
nav_after = None
nav_change = None
nav_before_list.append(nav_before)
nav_after_list.append(nav_after)
nav_change_list.append(nav_change)
# 添加净值信息列
rebalance_df['调仓前净值'] = nav_before_list
rebalance_df['调仓后净值'] = nav_after_list
rebalance_df['净值变化%'] = nav_change_list
return rebalance_df
def _calculate_net_value(self, result: pd.DataFrame) -> pd.DataFrame:
"""计算净值（起点归一化）"""
result['策略净值'] = (1 + result['策略日收益率']).cumprod()