diff --git a/scripts/strategy_lifecycle.py b/scripts/strategy_lifecycle.py index 343e4d5d..c8d61a95 100644 --- a/scripts/strategy_lifecycle.py +++ b/scripts/strategy_lifecycle.py @@ -2528,5 +2528,150 @@ def regenerate_all(stdout=True): 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__": - 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()