#!/usr/bin/env python3 # -*- coding: utf-8 -*- """swap_decision.py — 换仓决策模块（2026-08-18 老莫：资金腾挪时对比预期收益取最低者卖） 核心：需要资金买入新票时，对比【新票预期收益 E_new】vs【持仓不卖预期收益 E_hold】， 按 E_hold 升序选票卖出凑钱（卖预期收益最低的）。能代码算的尽量代码算，LLM 只做定性修正。 数据依据（10y 回测实测，2026-08-18 验证）： E_hold 查表（深套等到底均收益）: -20~-25%: -3.8% (30d恢复76% 60d85% 120d92%) -25~-30%: -3.7% (30d55% 60d72% 120d85%) -30~-40%: -2.9% (30d29% 60d38% 120d65%) -40%以下: -0.3% (30d18% 60d21% 120d48%) E_new 查表（策略avg_pnl，strategy_regime_perf_by_period 温区级优先，fallback 10y整体）: v_weak: +3.34% | b_td1_v3: +8.01% | s2_panic: +16.61% | v_next/v8.1: +9.39% 安全边际: E_new - E_hold >= 2%（覆盖双边交易成本~0.4% + 新票亏损风险） """ import sqlite3, json DB = "/home/hmo/MoFin/data/mofin.db" SAFE_MARGIN = 2.0 # 换仓安全边际（%） # 深套深度 → E_hold（等到底均收益%）— 2026-08-18 10y回测实测 E_HOLD_TABLE = [ # (dd_min, dd_max, e_hold_pct, note) (-100, -40, -0.3, "深套>40%：弹回概率极低，等到底期望≈-0.3%"), (-40, -30, -2.9, "深套30-40%：120天仅65%恢复，期望-2.9%"), (-30, -25, -3.7, "深套25-30%：60天72%恢复，期望-3.7%"), (-25, -20, -3.8, "深套20-25%：30天76%恢复，期望-3.8%"), ] def _get_conn(): conn = sqlite3.connect(DB, timeout=30) conn.execute("PRAGMA busy_timeout=30000") return conn def get_current_regime(market="a"): """当前温区""" try: conn = _get_conn() r = conn.execute("SELECT regime FROM market_regime WHERE market=? ORDER BY date DESC LIMIT 1", (market,)).fetchone() conn.close() return r[0] if r else "unknown" except Exception: return "unknown" def get_strategy_expected(strategy, market="a", regime=None): """E_new：策略预期单笔收益。温区级优先（strategy_regime_perf_by_period），fallback 10y整体（strategy_research）。 返回 (avg_pnl, source_desc)""" conn = _get_conn() try: # 温区级优先 if regime and regime != "unknown": r = conn.execute( "SELECT avg_pnl, trades FROM strategy_regime_perf_by_period " "WHERE strategy=? AND market=? AND regime=? AND period_tag='2y'", (strategy, market, regime)).fetchone() if r and r[0] is not None and r[1] and r[1] >= 5: conn.close() return (float(r[0]), f"温区{regime}2y({r[1]}笔)") # fallback 10y 整体 r2 = conn.execute( "SELECT results_json FROM strategy_research WHERE version=? AND COALESCE(market,'a')=? AND period_tag='10y' " "ORDER BY id DESC LIMIT 1", (strategy, market)).fetchone() if r2 and r2[0]: trades = json.loads(r2[0]).get("trades", []) if trades: avg = sum(t.get("profit_pct", 0) for t in trades) / len(trades) conn.close() return (round(avg, 2), f"10y整体({len(trades)}笔)") except Exception: pass conn.close() return (0.0, "无数据") def get_hold_expected(dd_pct): """E_hold：深套持仓等到底期望。按深套深度查表。非深套（dd>-20）返回 None（不算深套）。""" if dd_pct is None or dd_pct > -20: return None for lo, hi, e_hold, note in E_HOLD_TABLE: if lo <= dd_pct < hi: return {"e_hold": e_hold, "note": note, "dd": round(dd_pct, 1)} return {"e_hold": E_HOLD_TABLE[-1][2], "note": E_HOLD_TABLE[-1][3], "dd": round(dd_pct, 1)} def compute_hold_pnl(cost, price): """持仓浮盈%（成本 vs 现价）""" if not cost or not price: return None return (price - cost) / cost * 100 def decide_swap(need_cash, holdings, new_strategy, new_expected=None, market="a", regime=None): """核心换仓决策。 need_cash: 需要腾出的资金（元） holdings: [{code, name, cost, price, shares, strategy_name, dd_pct(深套深度，可选), pnl_pct}] new_strategy: 新标的的策略名 new_expected: 新标的预期收益（外部已算），None 则查表 返回: {decided, sell_list, need_cash, raised, reason, e_new, e_new_src} """ regime = regime or get_current_regime(market) # E_new if new_expected is None: e_new, e_new_src = get_strategy_expected(new_strategy, market, regime) else: e_new, e_new_src = new_expected, "外部提供" # 每只持仓算 E_hold scored = [] for h in holdings: code = h.get("code", "") name = h.get("name", code) cost = h.get("cost") or 0 price = h.get("price") or 0 shares = h.get("shares") or 0 market_val = price * shares if price and shares else 0 # 深套判定：优先用外部给的 dd_pct，否则用浮盈算（-20% 以下 = 深套） dd = h.get("dd_pct") pnl = h.get("pnl_pct") if dd is None and pnl is None and cost and price: pnl = compute_hold_pnl(cost, price) if dd is None: dd = pnl # 浮盈为负即深套深度近似 eh = get_hold_expected(dd) if (dd is not None and dd <= -20) else None # 非深套持仓：E_hold = 其自身策略的预期（继续持有的期望） h_strategy = h.get("strategy_name") or "" if eh is None: e_self, src_self = get_strategy_expected(h_strategy, market, regime) if h_strategy else (0.0, "无策略") eh = {"e_hold": e_self, "note": f"非深套，按原策略{h_strategy or 'unknown'}期望", "dd": None} scored.append({ "code": code, "name": name, "market_val": market_val, "e_hold": eh["e_hold"], "note": eh["note"], "dd": eh["dd"], "pnl": pnl, }) # 按 E_hold 升序（最低优先卖） scored.sort(key=lambda x: x["e_hold"]) # 累加凑钱 sell_list = [] raised = 0.0 for s in scored: if raised >= need_cash: break if s["market_val"] <= 0: continue sell_list.append(s) raised += s["market_val"] # 决策 if not sell_list: return {"decided": False, "reason": "无可卖持仓", "sell_list": [], "raised": 0, "e_new": e_new} # 安全边际：被卖的最后一只 E_hold vs E_new last_ehold = sell_list[-1]["e_hold"] margin = e_new - last_ehold if margin >= SAFE_MARGIN: decided = True reason = (f"换仓: 新票({new_strategy})E={e_new:.1f}%[{e_new_src}] " f"vs 被卖最后一只E={last_ehold:.1f}%({sell_list[-1]['name']}), 边际{margin:.1f}%≥{SAFE_MARGIN}%") else: decided = False reason = (f"不换: 新票({new_strategy})E={e_new:.1f}% vs 最低E={last_ehold:.1f}%, " f"边际{margin:.1f}%<{SAFE_MARGIN}%（不划算）") return { "decided": decided, "reason": reason, "sell_list": sell_list, "raised": round(raised, 0), "need_cash": need_cash, "e_new": e_new, "e_new_src": e_new_src, "margin": round(margin, 1), "regime": regime, } def format_swap_advice(decision): """格式化换仓建议（供 LLM prompt 注入 / XMPP 推送）""" if not decision.get("decided"): return f"【换仓决策】{decision.get('reason')}" lines = [f"【换仓决策】{decision.get('reason')}"] lines.append(f" 需资金 {decision.get('need_cash',0):.0f}元，卖出 {len(decision['sell_list'])} 只（共{decision.get('raised',0):.0f}元）:") for s in decision["sell_list"]: lines.append(f" - {s['code']} {s['name']}: E_hold={s['e_hold']:.1f}% {s['note']}") return "\n".join(lines) if __name__ == "__main__": # 自测 holdings = [ {"code": "000850", "name": "华茂", "cost": 3.84, "price": 4.14, "shares": 30600, "strategy_name": ""}, {"code": "688775", "name": "影石", "cost": 130, "price": 126.27, "shares": 100, "strategy_name": "accumulation"}, {"code": "300750", "name": "宁德", "cost": 500, "price": 393, "shares": 100, "strategy_name": ""}, ] d = decide_swap(need_cash=100000, holdings=holdings, new_strategy="b_td1_v3", market="a") print(format_swap_advice(d)) print(json.dumps(d, ensure_ascii=False, indent=1)[:800])