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

feat(rotation): 增加最新调仓信号展示功能

Browse Source 
- 配置中取消固定end_date，改为默认使用当前日期
- 添加打印最新调仓信号的功能，显示持仓明细及调出品种
- 在报告生成流程中调用最新调仓信号打印函数
- 图表展示中新增最新调仓信号表格，支持颜色区分调入、调出和维持
- 优化报告图表布局，调整画布高度适应信号表内容
- 删除无用test.py测试脚本及相关冗余代码
This commit is contained in:
aszerW
2026-03-19 00:22:25 +08:00
parent 02ba6a5df2
commit 9b154a1a25
4 changed files with 262 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

260
strategies/rotation/report.py
View File

@@ -77,6 +77,9 @@ def generate_performance_report(
print(f' {"最大回撤区间":<22} {str(s_dd_start.date()):>10} ~ {str(s_dd_end.date())}')
print("=" * 70)
# 打印最新调仓信号
_print_latest_signal(backtest_result, code_list, code_name_map, select_num)
# 绘制图表
_plot_report_chart(
backtest_result, code_list, code_name_map,
@@ -99,6 +102,178 @@ def generate_performance_report(
}
def _print_latest_signal(backtest_result: pd.DataFrame, code_list: list, code_name_map: dict, select_num: int):
"""打印最新调仓信号"""
latest = _extract_latest_positions(backtest_result, code_list, code_name_map, select_num)
signal_date_str = latest["signal_date"].strftime("%Y-%m-%d")
print("\n")
print("=" * 100)
print(" 最新调仓信号 (下一交易日执行)")
print("=" * 100)
print(f" 数据截止: {signal_date_str}")
print()
# 表头
print(f' {"品种名称":<8} {"代码":>10} {"仓位":>6} {"得分":>8} {"进场日期":>12} {"进场价":>10} {"最新价":>10} {"操作":>6} {"持有天数":>8} {"盈亏":>10}')
print(" " + "-" * 115)
# 下期持仓（调入/维持）
for pos in latest["positions"]:
pnl_str = f'{pos["pnl"]:>+9.2%}' if pos["pnl"] is not None else ' —'
days_str = f'{pos["holding_days"]:>7}天' if pos["holding_days"] is not None else ' —'
entry_str = f'{pos["entry_price"]:>10.2f}' if pos["entry_price"] is not None else ' —'
entry_date_str = pos["entry_date"].strftime("%Y-%m-%d") if pos.get("entry_date") else ' —'
score_str = f'{pos["score"]:>8.2f}' if pos["score"] is not None else ' —'
flag = '▲' if pos["action"] == "调入" else ' '
print(f' {pos["name"]:<8} {pos["code"]:>10} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} {flag}{pos["action"]:>4} {days_str} {pnl_str}')
# 需调出的品种
if latest["exit_positions"]:
print()
print(" 需调出:")
for pos in latest["exit_positions"]:
pnl_str = f'{pos["pnl"]:>+9.2%}' if pos["pnl"] is not None else ' —'
days_str = f'{pos["holding_days"]:>7}天' if pos["holding_days"] is not None else ' —'
entry_str = f'{pos["entry_price"]:>10.2f}' if pos["entry_price"] is not None else ' —'
entry_date_str = pos["entry_date"].strftime("%Y-%m-%d") if pos.get("entry_date") else ' —'
score_str = ' —' # 调出品种无得分
print(f' {pos["name"]:<8} {pos["code"]:>10} {pos["weight"]:>6.0%} {score_str} {entry_date_str:>12} {entry_str} {pos["current_price"]:>10.2f} ▼调出 {days_str} {pnl_str}')
print("=" * 120)
def _extract_latest_positions(backtest_result: pd.DataFrame, code_list: list, code_name_map: dict, select_num: int) -> dict:
"""提取最新持仓和下期调仓建议"""
last_date = backtest_result.index[-1]
last_row = backtest_result.iloc[-1]
# 当前持仓
current_signal = last_row["信号"]
if select_num == 1:
current_codes = [current_signal]
else:
current_codes = current_signal.split(",")
# 下期建议
score_cols = [f"得分_{code}" for code in code_list if f"得分_{code}" in backtest_result.columns]
scores = pd.to_numeric(last_row[score_cols], errors="coerce")
top_n = scores.nlargest(select_num)
next_codes = [c.replace("得分_", "") for c in top_n.index]
# 计算持仓信息
positions_info = []
weight = 1.0 / select_num
for code in next_codes:
name = code_name_map.get(code, code)
action = "维持" if code in current_codes else "调入"
# 获取当前价格和得分
current_price = last_row.get(code, 0)
score = scores.get(f"得分_{code}", None)
# 计算持仓信息（如果是维持的仓位）
entry_date = None
entry_price = None
holding_days = None
pnl = None
if action == "维持":
# 找到该标的最近一次连续持仓的起始日期
signal_series = backtest_result["信号"]
mask = signal_series == code if select_num == 1 else signal_series.str.contains(code, regex=False, na=False)
# 找到连续持仓段（从后往前找）
is_holding = mask.values
dates = backtest_result.index
# 从最后一天往前遍历，找到连续持仓的起始点
entry_date = None
for i in range(len(is_holding) - 1, -1, -1):
if is_holding[i]:
entry_date = dates[i]
else:
break
if entry_date is not None:
entry_price = backtest_result.loc[entry_date, code]
holding_days = (last_date - entry_date).days
if entry_price and entry_price != 0:
pnl = current_price / entry_price - 1
positions_info.append({
"code": code,
"name": name,
"weight": weight,
"score": score,
"action": action,
"current_price": current_price,
"entry_date": entry_date,
"entry_price": entry_price,
"holding_days": holding_days,
"pnl": pnl,
})
# 需调出的品种信息
exit_positions = []
for code in current_codes:
if code not in next_codes:
name = code_name_map.get(code, code)
current_price = last_row.get(code, 0)
# 计算调出品种的持仓信息（最近一次连续持仓）
signal_series = backtest_result["信号"]
mask = signal_series == code if select_num == 1 else signal_series.str.contains(code, regex=False, na=False)
# 找到连续持仓段（从后往前找）
is_holding = mask.values
dates = backtest_result.index
entry_price = None
holding_days = None
pnl = None
# 从最后一天往前遍历，找到连续持仓的起始点
entry_date = None
for i in range(len(is_holding) - 1, -1, -1):
if is_holding[i]:
entry_date = dates[i]
else:
break
if entry_date is not None:
entry_price = backtest_result.loc[entry_date, code]
holding_days = (last_date - entry_date).days
if entry_price and entry_price != 0:
pnl = current_price / entry_price - 1
exit_positions.append({
"code": code,
"name": name,
"weight": weight,
"score": None, # 调出品种无得分
"action": "调出",
"current_price": current_price,
"entry_date": entry_date,
"entry_price": entry_price,
"holding_days": holding_days,
"pnl": pnl,
})
return {
"signal_date": last_date,
"current_codes": current_codes,
"next_codes": next_codes,
"positions": positions_info,
"exit_positions": exit_positions,
}
def _plot_report_chart(
backtest_result: pd.DataFrame,
code_list: list,
@@ -114,10 +289,86 @@ def _plot_report_chart(
strategy_nav = backtest_result["轮动策略净值"]
benchmark_nav = backtest_result["基准净值"]
fig, axes = plt.subplots(3, 1, figsize=(14, 12))
# 提取最新调仓信息
latest = _extract_latest_positions(backtest_result, code_list, code_name_map, select_num)
# 计算表格行数
n_table_rows = len(latest["positions"]) + len(latest["exit_positions"])
table_height = max(1.5, 0.5 + n_table_rows * 0.28)
fig = plt.figure(figsize=(14, 10 + table_height + 8))
gs = fig.add_gridspec(4, 1, height_ratios=[table_height, 3, 1, 1.2], hspace=0.35)
# 面板0: 最新调仓信号表
ax0 = fig.add_subplot(gs[0])
ax0.axis("off")
signal_date_str = latest["signal_date"].strftime("%Y-%m-%d")
ax0.set_title(f"最新调仓信号 (数据截止: {signal_date_str}，下一交易日执行)", fontsize=14, fontweight="bold", loc="left", pad=15)
# 构建表格数据
table_data = []
col_labels = ["品种名称", "代码", "仓位", "得分", "进场日期", "进场价", "最新价", "操作", "持有天数", "盈亏"]
# 下期持仓（调入/维持）
for pos in latest["positions"]:
pnl_str = f'{pos["pnl"]:+.2%}' if pos["pnl"] is not None else "—"
days_str = f'{pos["holding_days"]}' if pos["holding_days"] is not None else "—"
entry_str = f'{pos["entry_price"]:.2f}' if pos["entry_price"] is not None else "—"
entry_date_str = pos["entry_date"].strftime("%m-%d") if pos.get("entry_date") else "—"
score_str = f'{pos["score"]:.2f}' if pos["score"] is not None else "—"
table_data.append([
pos["name"], pos["code"], f'{pos["weight"]:.0%}',
score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}',
pos["action"], days_str, pnl_str
])
# 需调出的品种
for pos in latest["exit_positions"]:
pnl_str = f'{pos["pnl"]:+.2%}' if pos["pnl"] is not None else "—"
days_str = f'{pos["holding_days"]}' if pos["holding_days"] is not None else "—"
entry_str = f'{pos["entry_price"]:.2f}' if pos["entry_price"] is not None else "—"
entry_date_str = pos["entry_date"].strftime("%m-%d") if pos.get("entry_date") else "—"
score_str = "—" # 调出品种无得分
table_data.append([
pos["name"], pos["code"], f'{pos["weight"]:.0%}',
score_str, entry_date_str, entry_str, f'{pos["current_price"]:.2f}',
"调出", days_str, pnl_str
])
if table_data:
table = ax0.table(
cellText=table_data,
colLabels=col_labels,
loc="upper center",
cellLoc="center",
colWidths=[0.08, 0.08, 0.05, 0.06, 0.06, 0.07, 0.07, 0.05, 0.06, 0.07],
)
table.auto_set_font_size(False)
table.set_fontsize(10)
table.scale(1, 2.0)
# 表头深色
for j in range(len(col_labels)):
table[0, j].set_facecolor("#2C3E50")
table[0, j].set_text_props(color="white", fontweight="bold")
# 数据行按操作着色
for i in range(len(table_data)):
action = table_data[i][3]
if action == "调入":
color = "#d4edda" # 绿色
elif action == "调出":
color = "#f8d7da" # 红色
else:
color = "#fff3cd" # 黄色
for j in range(len(col_labels)):
table[i + 1, j].set_facecolor(color)
# 面板1: 净值曲线
ax1 = axes[0]
ax1 = fig.add_subplot(gs[1])
ax1.plot(strategy_nav.index, strategy_nav.values,
label="轮动策略", linewidth=2, color="#E74C3C")
ax1.plot(benchmark_nav.index, benchmark_nav.values,
@@ -139,7 +390,7 @@ def _plot_report_chart(
ax1.set_yscale("log")
# 面板2: 回撤曲线
ax2 = axes[1]
ax2 = fig.add_subplot(gs[2])
cummax = strategy_nav.cummax()
drawdown = (strategy_nav - cummax) / cummax
ax2.fill_between(drawdown.index, drawdown.values, 0, alpha=0.5, color="#E74C3C")
@@ -148,7 +399,7 @@ def _plot_report_chart(
ax2.grid(True, alpha=0.3)
# 面板3: 持仓分布
ax3 = axes[2]
ax3 = fig.add_subplot(gs[3])
signal_series = backtest_result["信号"]
for i, code in enumerate(code_list):
name = code_name_map.get(code, code)
@@ -168,7 +419,6 @@ def _plot_report_chart(
ax3.set_yticklabels(ylabels, fontsize=7)
ax3.grid(True, alpha=0.3)
plt.tight_layout()
chart_path = f"{save_path}_chart.png"
plt.savefig(chart_path, dpi=150, bbox_inches="tight")
plt.close()