feat: 换股规划 swap-plan — 深套股可替代性排序+资金释放方案

- print_swap_plan(): 五维评分(亏损程度/仓位/技术趋势/流动性/前景)
- 排序输出,明确标注"当前亏损X%深套"而非"深套持有"
- --need 参数模拟指定资金需求,自动计算卖几只够
- 可独立运行:python strategy_lifecycle.py swap-plan [--need 金额]
This commit is contained in:
知微
2026-07-13 12:05:36 +08:00
parent a6d37c6f30
commit 2109289668
+146 -1
View File
@@ -2528,5 +2528,150 @@ def regenerate_all(stdout=True):
return summary return summary
# ── 换股规划(调仓优先级排序) ──
def print_swap_plan(need_cash: float = 0):
"""输出深套股调仓优先级排序
Args:
need_cash: 需要的资金额(元),0=显示全部排序不指定金额
"""
import sqlite3
from mo_data import read_portfolio
from mofin_db import get_conn
pf = read_portfolio()
holdings = pf.get("holdings", [])
if not holdings:
print("无持仓数据")
return
total_mv = pf.get("total_mv", 0)
total_cash = pf.get("cash_available", pf.get("cash", 0))
print(f"📊 当前总市值={total_mv:,.0f} 可用现金={total_cash:,.0f}")
if need_cash > 0:
print(f"🎯 需要释放资金: {need_cash:,.0f}")
print()
# 计算每只深套股的评分
scored = []
for h in holdings:
code = h["code"]
name = h["name"]
price = h.get("price", 0)
cost = h.get("cost", 0)
shares = h.get("shares", 0)
mv = h.get("market_value", 0) or shares * price
pnl_pct = (price - cost) / cost * 100 if cost > 0 else 0
position_pct = h.get("position_pct", mv / total_mv * 100 if total_mv > 0 else 0)
currency = h.get("currency", "CNY")
is_hk = len(str(code)) == 5 and str(code)[0] in ("0", "1")
# 只评估亏损股(含微亏)
if pnl_pct >= 0:
continue
# ── 可替代性评分(0~100,越高越建议优先替换) ──
# ① 亏损程度(越低越容易割,权重25%)
abs_loss = abs(pnl_pct)
loss_score = max(0, 100 - (abs_loss - 5) * 1.33) # -5%→100, -20%→80, -50%→40, -80%→0
# ② 仓位市值(越大释放资金越多,权重20%)
mv_pct = position_pct
size_score = min(100, mv_pct * 5) # 5%仓位→25分, 20%→100分
# ③ 技术面趋势(权重25%)- 从 DB 取趋势判断
trend_score = 50 # 默认中性
try:
conn = get_conn()
r = conn.execute(
"SELECT timing_signal FROM holding_strategies WHERE code=? AND status='active' ORDER BY updated_at DESC LIMIT 1",
(code,)
).fetchone()
conn.close()
if r:
sig = r[0] or ""
if any(kw in sig for kw in ["卖出", "止损", "离场", "看空"]):
trend_score = 90
elif any(kw in sig for kw in ["买入", "加仓", "看多", "走强"]):
trend_score = 10
elif any(kw in sig for kw in ["观望", "信号不充分", "中性"]):
trend_score = 60
except:
pass
# ④ 流动性(权重15%)- H股略打折扣
liquidity_score = 60 if is_hk else 80
# ⑤ 综合评分
total_score = (
loss_score * 0.25 +
size_score * 0.20 +
trend_score * 0.25 +
liquidity_score * 0.15 +
50 * 0.15 # 前景占15%,暂给中性50分
)
scored.append({
"code": code, "name": name,
"price": price, "pnl_pct": round(pnl_pct, 1),
"mv": mv, "mv_pct": round(mv_pct, 1),
"score": round(total_score, 1),
"currency": currency,
"loss_score": round(loss_score, 0),
"trend_score": round(trend_score, 0),
"shares": shares,
})
if not scored:
print("当前无亏损持仓,无需调仓规划")
return
# 按评分降序排列(越高越优先替换)
scored.sort(key=lambda s: -s["score"])
print(f"{'优先级':>4} {'代码':<8} {'名称':<14} {'亏损':>6} {'市值':>10} {'占比':>5} {'评分':>4} 理由")
print("-" * 90)
total_releasable = 0
for i, s in enumerate(scored, 1):
currency_tag = "HK$" if s["currency"] == "HKD" else "CNY"
reasons = []
if s["loss_score"] >= 70:
reasons.append("亏损较浅易割")
elif s["loss_score"] <= 30:
reasons.append("亏损较深可等反弹")
if s["trend_score"] >= 70:
reasons.append("技术弱势")
elif s["trend_score"] <= 30:
reasons.append("技术尚可")
if s["mv_pct"] >= 10:
reasons.append("仓位较重释放资金多")
reason_str = "".join(reasons[:2]) if reasons else "中性"
print(f" #{i:<2} {s['code']:<8} {s['name']:<12} {s['pnl_pct']:>5.1f}% {currency_tag}{s['mv']:>8,.0f} {s['mv_pct']:>4.1f}% {s['score']:>4.1f} {reason_str}")
if need_cash > 0 and total_releasable < need_cash:
total_releasable += s["mv"]
if total_releasable >= need_cash:
print(f"\n ▶ 卖出前{i}只可释放 {currency_tag}{total_releasable:,.0f}(目标{need_cash:,.0f}")
print()
print("说明:评分越高 = 越优先考虑替换。综合亏损程度/仓位/技术趋势/流动性。")
print("评分仅供参考,最终决策请结合当日行情和个人判断。")
print(f"\n若需模拟换股方案执行: python3 {__file__} swap-plan --need 金额")
if __name__ == "__main__": if __name__ == "__main__":
regenerate_all() import sys
if len(sys.argv) > 1 and sys.argv[1] == "swap-plan":
need = 0
for i, a in enumerate(sys.argv):
if a == "--need" and i + 1 < len(sys.argv):
try:
need = float(sys.argv[i + 1])
except:
pass
print_swap_plan(need)
else:
regenerate_all()