feat: 换仓决策模块——资金不足时E_new(新票预期)vs E_hold(持仓深套查表)对比,卖E最低的凑钱(老莫:能代码算的不让LLM拍脑袋)。重评注入换仓方案
This commit is contained in:
xxm
@@ -412,6 +412,51 @@ def build_prompt(data):
|
||||
except Exception:
|
||||
pass
|
||||
_position_context += _rotation_context
|
||||
# 2026-08-18 换仓决策注入（老莫：需要资金时对比预期收益，卖E_hold最低的）
|
||||
# 触发：本票信号为买入/加仓 + 现金不足（cash < 建议仓位金额估算）
|
||||
try:
|
||||
from swap_decision import decide_swap, format_swap_advice, get_strategy_expected
|
||||
_sig_now = data.get("timing_signal") or ""
|
||||
_need_fund = _sig_now in ("买入", "可买入", "可加仓")
|
||||
if _need_fund:
|
||||
# 估算本票建议仓位金额（按 RR 5%-15%，取中 10%）
|
||||
_est_need = total * 0.10
|
||||
if cash < _est_need:
|
||||
# 拉当前持仓（深套判定：cost vs price）
|
||||
try:
|
||||
import sqlite3 as _sq3
|
||||
_c3 = _sq3.connect(DB, timeout=30)
|
||||
_c3.row_factory = _sq3.Row
|
||||
_holds = _c3.execute("SELECT code, name, shares, cost, price FROM holdings WHERE is_active=1 AND shares>0").fetchall()
|
||||
_c3.close()
|
||||
_hold_list = []
|
||||
for _h in _holds:
|
||||
_cost_v = _h["cost"] or 0; _price_v = _h["price"] or 0
|
||||
_pnl_v = (_price_v - _cost_v) / _cost_v * 100 if _cost_v > 0 and _price_v > 0 else None
|
||||
# 查该持仓的 strategy_name
|
||||
_sn = ""
|
||||
try:
|
||||
_c4 = _sq3.connect(DB, timeout=30)
|
||||
_sn_r = _c4.execute("SELECT strategy_name FROM holding_strategies WHERE code=? AND status='active'", (_h["code"],)).fetchone()
|
||||
_c4.close()
|
||||
_sn = _sn_r[0] if _sn_r else ""
|
||||
except Exception:
|
||||
pass
|
||||
_hold_list.append({
|
||||
"code": _h["code"], "name": _h["name"] or _h["code"],
|
||||
"cost": _cost_v, "price": _price_v, "shares": _h["shares"] or 0,
|
||||
"strategy_name": _sn, "pnl_pct": _pnl_v,
|
||||
})
|
||||
_sd = data.get("strategy_name") or data.get("strategy_attributed") or ""
|
||||
_dec = decide_swap(need_cash=_est_need - cash, holdings=_hold_list,
|
||||
new_strategy=_sd, market="a")
|
||||
_swap_advice = format_swap_advice(_dec)
|
||||
_position_context += f"\n\n{_swap_advice}\n⚠️ 若你给出买入建议但现金不足，以上为系统算好的换仓方案（卖预期收益最低的持仓凑钱）；你只需确认是否采纳（可结合消息面/基本面修正），不要重新设计换仓逻辑。"
|
||||
print(f" [SWAP] 换仓决策已注入: {_dec.get('reason', '')[:80]}", flush=True)
|
||||
except Exception as _se:
|
||||
print(f" [SWAP] 换仓决策注入失败: {_se}", flush=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── 策略语义注入（2026-08-18 老莫：重评按策略定义，不是裸标签）──
|
||||
_sd = data.get("strategy_def")
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
#!/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])
|
||||
Reference in New Issue
Block a user