""" 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}")