185 lines
8.3 KiB
Python
185 lines
8.3 KiB
Python
#!/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])
|