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

feat(report): 全标的动量排名表替代原调仓信号表

Browse Source 
Panel 0 从仅展示持仓调仓扩展为全标的排名表：
- 新增未入选标的行，按动量降序展示
- 新增排名、市场两列
- 表格按调入→维持→调出→未入选顺序排列
- 调出标的也展示真实得分（便于分析调出原因）
- 标题显示当前动态阈值
- 未入选标的浅灰背景区分
This commit is contained in:
aszerW
2026-06-08 00:12:17 +08:00
parent 6a5ae8efbf
commit 13c69c2a0b
1 changed files with 57 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
rotation/simple_rotation.py
View File

@@ -1256,11 +1256,9 @@ class SimpleRotationStrategy:
'holding_days': holding_days, 'pnl': pnl,
})
# Find exit positions: ONLY show when the last day is a rebalance day
# If last day is NOT rebalance, don't show any "调出" info (already past)
# Build exit positions: ONLY when the last day is a rebalance day
exit_positions = []
if last_rec.get('is_rebalance', False):
# Find the previous day's holdings to compare
last_idx = len(self.daily_records) - 1
if last_idx > 0:
old_holdings = set(self.daily_records[last_idx - 1]['holdings'])
@@ -1274,71 +1272,100 @@ class SimpleRotationStrategy:
trade_code = self.signal_to_trade.get(code, code)
etf_close = self._get_etf_close(trade_code, last_date)
premium = self._get_latest_premium(trade_code, last_date)
score = factors.get(code)
exit_positions.append({
'name': name, 'code': code, 'etf': etf_code,
'weight': 0, 'score': None,
'weight': 0, 'score': score,
'idx_close': idx_close, 'etf_close': etf_close,
'premium': premium, 'action': '调出',
'entry_date': None, 'entry_price': None,
'holding_days': 0, 'pnl': None,
})
# Build unselected (未入选) positions: all signal codes not held and not exited
held_or_exited = set(holdings) | {p['code'] for p in exit_positions}
unselected_positions = []
# Sort unselected by momentum descending
unselected_codes = [c for c in self.signal_codes if c not in held_or_exited]
unselected_codes.sort(key=lambda c: factors.get(c, -999), reverse=True)
for code in unselected_codes:
cfg = code_config.get(code, {})
name = cfg.get('name', code)
etf_code = cfg.get('etf', '—')
score = factors.get(code)
idx_close = self._get_index_close(code, last_date)
trade_code = self.signal_to_trade.get(code, code)
etf_close = self._get_etf_close(trade_code, last_date)
premium = self._get_latest_premium(trade_code, last_date)
unselected_positions.append({
'name': name, 'code': code, 'etf': etf_code,
'weight': 0, 'score': score,
'idx_close': idx_close, 'etf_close': etf_close,
'premium': premium, 'action': '未入选',
'entry_date': None, 'entry_price': None,
'holding_days': 0, 'pnl': None,
})
# ==================== Plot ====================
plt.rcParams["font.sans-serif"] = ["Arial Unicode MS", "WenQuanYi Zen Hei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
n_rows = len(positions_info) + len(exit_positions)
n_rows = len(positions_info) + len(exit_positions) + len(unselected_positions)
signal_h = max(1.5, 0.5 + n_rows * 0.35)
fig = plt.figure(figsize=(16, 10 + signal_h + 1.2 + 8))
gs = fig.add_gridspec(5, 1, height_ratios=[signal_h, 1.2, 3, 1, 1.2], hspace=0.35)
# Panel 0: Signal table
# Panel 0: Full asset ranking table
ax0 = fig.add_subplot(gs[0])
ax0.axis("off")
ax0.set_title(f"最新调仓信号 (信号日期: {last_date.strftime('%Y-%m-%d')}，下一交易日执行)",
threshold_val = last_rec.get('threshold', 0.0)
ax0.set_title(f"全标的动量排名 (信号日期: {last_date.strftime('%Y-%m-%d')}，阈值: {threshold_val:.4f})",
fontsize=14, fontweight="bold", loc="left", pad=15)
col_labels = ["标的名称", "指数代码", "ETF代码", "仓位", "得分", "指数最新价",
"ETF收盘价", "溢价率", "操作", "持有天数", "盈亏"]
# Build rank map for all codes by momentum
ranked_all = sorted(factors.keys(), key=lambda c: factors[c], reverse=True)
rank_map = {c: i + 1 for i, c in enumerate(ranked_all)}
col_labels = ["排名", "标的名称", "市场", "指数代码", "ETF代码", "仓位", "得分",
"指数最新价", "ETF收盘价", "溢价率", "状态", "持有天数", "盈亏"]
table_data = []
for p in positions_info:
score_s = f"{p['score']:.2f}" if p['score'] is not None else "—"
row_actions = [] # track action for coloring
# Ordered: 调入 -> 维持 -> 调出 -> 未入选
all_rows = positions_info + exit_positions + unselected_positions
for p in all_rows:
rank = rank_map.get(p['code'], '—')
score_s = f"{p['score']:.4f}" if p['score'] is not None else "—"
idx_s = f"{p['idx_close']:.2f}" if p['idx_close'] is not None else "—"
etf_s = f"{p['etf_close']:.3f}" if p['etf_close'] is not None else "—"
prem_s = f"{p['premium']:+.2%}" if p['premium'] is not None else "—"
days_s = str(p['holding_days']) if p['holding_days'] > 0 else "—"
pnl_s = f"{p['pnl']:+.2%}" if p['pnl'] is not None else "—"
weight_s = f"{p['weight']:.0%}" if p['weight'] > 0 else "—"
market = code_config.get(p['code'], {}).get('market', '—')
table_data.append([
p['name'], p['code'], p['etf'], f"{p['weight']:.0%}",
rank, p['name'], market, p['code'], p['etf'], weight_s,
score_s, idx_s, etf_s, prem_s, p['action'], days_s, pnl_s
])
for p in exit_positions:
idx_s = f"{p['idx_close']:.2f}" if p['idx_close'] is not None else "—"
etf_s = f"{p['etf_close']:.3f}" if p['etf_close'] is not None else "—"
prem_s = f"{p['premium']:+.2%}" if p['premium'] is not None else "—"
table_data.append([
p['name'], p['code'], p['etf'], f"{p['weight']:.0%}",
"—", idx_s, etf_s, prem_s, "调出", "—", "—"
])
row_actions.append(p['action'])
if table_data:
tbl = ax0.table(cellText=table_data, colLabels=col_labels, loc="center",
cellLoc="center", bbox=[0, 0, 1, 1])
tbl.auto_set_font_size(False)
tbl.set_fontsize(9)
tbl.scale(1, 1.8)
tbl.set_fontsize(8)
tbl.scale(1, 1.6)
for j in range(len(col_labels)):
tbl[0, j].set_facecolor("#2C3E50")
tbl[0, j].set_text_props(color="white", fontweight="bold")
action_colors = {
"调入": "#d4edda", # 浅绿色
"调出": "#f8d7da", # 浅红色
"维持": "#fff3cd", # 浅黄色
"未入选": "#f0f0f0", # 浅灰色
}
for i in range(len(table_data)):
action = table_data[i][8]
# Legacy color scheme: 调入=green, 维持=yellow, 调出=red
if action == "调入":
color = "#d4edda" # 浅绿色
elif action == "调出":
color = "#f8d7da" # 浅红色
else:
color = "#fff3cd" # 浅黄色（维持）
color = action_colors.get(row_actions[i], "#ffffff")
for j in range(len(col_labels)):
tbl[i + 1, j].set_facecolor(color)