aszerW
70bb69fd98
- 修正CAGR计算，去除NaN并检查起始值有效性以避免异常结果 - 优化混合数据源的数据对齐逻辑，使用配置结束日期与A股最新数据日期的较早者 - 计算因子时对齐A股交易日历，重新基于对齐价格计算日收益率，改进因子对齐准确度 - 轮动策略中跳过空信号，避免空信号影响持仓和调仓逻辑 - 调整信号处理，过滤空字符串和NaN，保证轮动信号数据有效性 - 多品种轮动持仓中加入空信号判断，避免无效信号导致错误 - 调整调仓明细和品种汇总保存逻辑，增加空文件创建以保证输出路径文件稳定生成 - 完善多处打印信息和注释，增强代码可读性与调试便利性
253 lines
9.9 KiB
Python
253 lines
9.9 KiB
Python
"""
|
|
ETF轮动策略 - 持仓跟踪模块
|
|
"""
|
|
|
|
import pandas as pd
|
|
from typing import Optional
|
|
|
|
|
|
def track_positions(
|
|
backtest_result: pd.DataFrame,
|
|
code_name_map: dict = None,
|
|
select_num: int = 1,
|
|
) -> tuple[pd.DataFrame, pd.DataFrame]:
|
|
"""
|
|
从回测结果中提取每笔持仓记录
|
|
|
|
Args:
|
|
backtest_result: 回测结果（含 '信号' 列）
|
|
code_name_map: 代码→名称映射
|
|
select_num: 每次选中的品种数量
|
|
|
|
Returns:
|
|
tuple: (trades_df, summary_df)
|
|
"""
|
|
code_name_map = code_name_map or {}
|
|
data = backtest_result.copy()
|
|
dates = data.index.tolist()
|
|
signals = data["信号"].tolist()
|
|
trades = []
|
|
|
|
if select_num == 1:
|
|
# 单品种轮动
|
|
current_code = signals[0]
|
|
entry_date = dates[0]
|
|
entry_price = data.loc[entry_date, current_code]
|
|
entry_nav = data.loc[entry_date, "轮动策略净值"]
|
|
|
|
for i in range(1, len(dates)):
|
|
today_code = signals[i]
|
|
|
|
if today_code != current_code:
|
|
exit_date = dates[i - 1]
|
|
exit_price = data.loc[exit_date, current_code]
|
|
exit_nav = data.loc[exit_date, "轮动策略净值"]
|
|
holding_days = (i - 1) - dates.index(entry_date) + 1
|
|
trade_return = exit_price / entry_price - 1 if entry_price != 0 else 0
|
|
nav_contrib = exit_nav - entry_nav
|
|
|
|
trades.append({
|
|
"序号": len(trades) + 1,
|
|
"品种代码": current_code,
|
|
"品种名称": code_name_map.get(current_code, current_code),
|
|
"进场日期": entry_date,
|
|
"出场日期": exit_date,
|
|
"持仓天数": holding_days,
|
|
"仓位占比": "100%",
|
|
"进场价格": round(entry_price, 2),
|
|
"出场价格": round(exit_price, 2),
|
|
"持仓收益": trade_return,
|
|
"进场净值": round(entry_nav, 4),
|
|
"出场净值": round(exit_nav, 4),
|
|
"净值贡献": round(nav_contrib, 4),
|
|
})
|
|
|
|
current_code = today_code
|
|
entry_date = dates[i]
|
|
entry_price = data.loc[entry_date, current_code]
|
|
entry_nav = data.loc[entry_date, "轮动策略净值"]
|
|
|
|
# 最后一笔
|
|
exit_date = dates[-1]
|
|
exit_price = data.loc[exit_date, current_code]
|
|
exit_nav = data.loc[exit_date, "轮动策略净值"]
|
|
holding_days = len(dates) - dates.index(entry_date)
|
|
trade_return = exit_price / entry_price - 1 if entry_price != 0 else 0
|
|
nav_contrib = exit_nav - entry_nav
|
|
|
|
trades.append({
|
|
"序号": len(trades) + 1,
|
|
"品种代码": current_code,
|
|
"品种名称": code_name_map.get(current_code, current_code),
|
|
"进场日期": entry_date,
|
|
"出场日期": exit_date,
|
|
"持仓天数": holding_days,
|
|
"仓位占比": "100%",
|
|
"进场价格": round(entry_price, 2),
|
|
"出场价格": round(exit_price, 2),
|
|
"持仓收益": trade_return,
|
|
"进场净值": round(entry_nav, 4),
|
|
"出场净值": round(exit_nav, 4),
|
|
"净值贡献": round(nav_contrib, 4),
|
|
})
|
|
|
|
else:
|
|
# 多品种等权轮动
|
|
current_signal = signals[0]
|
|
entry_date = dates[0]
|
|
codes = [c for c in current_signal.split(",") if c] # 过滤空字符串
|
|
if not codes:
|
|
# 空信号，返回空结果
|
|
return pd.DataFrame(trades), pd.DataFrame()
|
|
weight = 1.0 / len(codes)
|
|
entry_prices = {c: data.loc[entry_date, c] for c in codes}
|
|
entry_nav = data.loc[entry_date, "轮动策略净值"]
|
|
|
|
for i in range(1, len(dates)):
|
|
today_signal = signals[i]
|
|
|
|
if today_signal != current_signal:
|
|
exit_date = dates[i - 1]
|
|
exit_nav = data.loc[exit_date, "轮动策略净值"]
|
|
holding_days = (i - 1) - dates.index(entry_date) + 1
|
|
|
|
for c in codes:
|
|
exit_price = data.loc[exit_date, c]
|
|
ep = entry_prices[c]
|
|
trade_return = exit_price / ep - 1 if ep != 0 else 0
|
|
|
|
trades.append({
|
|
"序号": len(trades) + 1,
|
|
"品种代码": c,
|
|
"品种名称": code_name_map.get(c, c),
|
|
"进场日期": entry_date,
|
|
"出场日期": exit_date,
|
|
"持仓天数": holding_days,
|
|
"仓位占比": f"{weight:.0%}",
|
|
"进场价格": round(ep, 2),
|
|
"出场价格": round(exit_price, 2),
|
|
"持仓收益": trade_return,
|
|
"进场净值": round(entry_nav, 4),
|
|
"出场净值": round(exit_nav, 4),
|
|
"净值贡献": round((exit_nav - entry_nav) * weight, 4),
|
|
})
|
|
|
|
current_signal = today_signal
|
|
entry_date = dates[i]
|
|
codes = [c for c in current_signal.split(",") if c] # 过滤空字符串
|
|
if not codes:
|
|
break # 空信号，结束循环
|
|
weight = 1.0 / len(codes)
|
|
entry_prices = {c: data.loc[entry_date, c] for c in codes}
|
|
entry_nav = data.loc[entry_date, "轮动策略净值"]
|
|
|
|
# 最后一笔
|
|
exit_date = dates[-1]
|
|
exit_nav = data.loc[exit_date, "轮动策略净值"]
|
|
holding_days = len(dates) - dates.index(entry_date)
|
|
for c in codes:
|
|
exit_price = data.loc[exit_date, c]
|
|
ep = entry_prices[c]
|
|
trade_return = exit_price / ep - 1 if ep != 0 else 0
|
|
trades.append({
|
|
"序号": len(trades) + 1,
|
|
"品种代码": c,
|
|
"品种名称": code_name_map.get(c, c),
|
|
"进场日期": entry_date,
|
|
"出场日期": exit_date,
|
|
"持仓天数": holding_days,
|
|
"仓位占比": f"{weight:.0%}",
|
|
"进场价格": round(ep, 2),
|
|
"出场价格": round(exit_price, 2),
|
|
"持仓收益": trade_return,
|
|
"进场净值": round(entry_nav, 4),
|
|
"出场净值": round(exit_nav, 4),
|
|
"净值贡献": round((exit_nav - entry_nav) * weight, 4),
|
|
})
|
|
|
|
trades_df = pd.DataFrame(trades)
|
|
summary = _summarize_by_code(trades_df, code_name_map)
|
|
return trades_df, summary
|
|
|
|
|
|
def _summarize_by_code(trades_df: pd.DataFrame, code_name_map: dict) -> pd.DataFrame:
|
|
"""按品种汇总持仓统计"""
|
|
if trades_df.empty:
|
|
return pd.DataFrame()
|
|
|
|
groups = trades_df.groupby("品种代码")
|
|
rows = []
|
|
|
|
for code, grp in groups:
|
|
total_trades = len(grp)
|
|
total_days = grp["持仓天数"].sum()
|
|
avg_days = grp["持仓天数"].mean()
|
|
win_trades = (grp["持仓收益"] > 0).sum()
|
|
win_rate = win_trades / total_trades if total_trades > 0 else 0
|
|
avg_return = grp["持仓收益"].mean()
|
|
total_return = (1 + grp["持仓收益"]).prod() - 1
|
|
max_return = grp["持仓收益"].max()
|
|
min_return = grp["持仓收益"].min()
|
|
|
|
rows.append({
|
|
"品种代码": code,
|
|
"品种名称": code_name_map.get(code, code),
|
|
"调仓次数": total_trades,
|
|
"总持仓天数": total_days,
|
|
"平均持仓天数": round(avg_days, 1),
|
|
"胜率": win_rate,
|
|
"平均收益": avg_return,
|
|
"累计收益": total_return,
|
|
"最大单次收益": max_return,
|
|
"最大单次亏损": min_return,
|
|
})
|
|
|
|
summary = pd.DataFrame(rows)
|
|
summary = summary.sort_values("总持仓天数", ascending=False).reset_index(drop=True)
|
|
return summary
|
|
|
|
|
|
def save_trades(
|
|
trades_df: pd.DataFrame,
|
|
summary_df: pd.DataFrame,
|
|
save_path: str = "report",
|
|
) -> None:
|
|
"""保存调仓明细和汇总到CSV"""
|
|
import os
|
|
os.makedirs(os.path.dirname(save_path) if os.path.dirname(save_path) else ".", exist_ok=True)
|
|
|
|
# 保存调仓明细
|
|
trades_path = f"{save_path}_trades.csv"
|
|
if not trades_df.empty:
|
|
trades_out = trades_df.copy()
|
|
if "持仓收益" in trades_out.columns:
|
|
trades_out["持仓收益"] = trades_out["持仓收益"].apply(lambda x: f"{x:.2%}")
|
|
if "进场日期" in trades_out.columns:
|
|
trades_out["进场日期"] = trades_out["进场日期"].apply(
|
|
lambda x: x.strftime("%Y-%m-%d") if hasattr(x, "strftime") else str(x)[:10]
|
|
)
|
|
if "出场日期" in trades_out.columns:
|
|
trades_out["出场日期"] = trades_out["出场日期"].apply(
|
|
lambda x: x.strftime("%Y-%m-%d") if hasattr(x, "strftime") else str(x)[:10]
|
|
)
|
|
trades_out.to_csv(trades_path, index=False, encoding="utf-8-sig")
|
|
print(f"\n调仓明细已保存: {trades_path}")
|
|
else:
|
|
# 创建空文件
|
|
pd.DataFrame().to_csv(trades_path, index=False, encoding="utf-8-sig")
|
|
print(f"\n调仓明细为空: {trades_path}")
|
|
|
|
# 保存品种汇总
|
|
summary_path = f"{save_path}_summary.csv"
|
|
if not summary_df.empty:
|
|
summary_out = summary_df.copy()
|
|
for col in ["胜率", "平均收益", "累计收益", "最大单次收益", "最大单次亏损"]:
|
|
if col in summary_out.columns:
|
|
summary_out[col] = summary_out[col].apply(lambda x: f"{x:.2%}")
|
|
summary_out.to_csv(summary_path, index=False, encoding="utf-8-sig")
|
|
print(f"品种汇总已保存: {summary_path}")
|
|
else:
|
|
# 创建空文件
|
|
pd.DataFrame().to_csv(summary_path, index=False, encoding="utf-8-sig")
|
|
print(f"品种汇总为空: {summary_path}")
|